[BUG] Report an Elasticsearch async export's outcome exactly once

AsyncResponseHandler called the result callback directly from OnResponse and from
each terminal OnEvent state with no guard, and ReadError, WriteError and Destroyed
fell through a default label and called nothing. The HTTP client can deliver both
a response and a terminal event for one request, so one export could report twice,
and it can end on one of those three states and report nothing at all.

The exporter counts one finished session per export. Reporting twice overshoots
that count for the life of the exporter. Reporting never leaves a flush waiting on
a session that has already ended.

Every path goes through one CompleteOnce now, a compare exchange that reports at
most once and keeps the first verdict. The switch lists every state with no
default, so a state added upstream fails to compile rather than going uncounted,
and the destructor reports a failure for a handler torn down without an outcome.

The completion line said trace span(s) in the log exporter and says log record(s)
now. The cases read that line to count outcomes, and the wording was wrong either
way.

Nine cases drive a fake HTTP client through the public constructor: each terminal
ordering a real session can produce, a response and a teardown event in both
orders, and the concurrent version of each. Removing the compare exchange turns
six of the nine red.

Extracted from #4337, which is 1526 lines and closes two issues. What stays there
is the ForceFlush deadline and watermark accounting for #4336, including four
completion cases that verify this guard through the flush rather than through the
log line.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

Author: thc1006

CHANGELOG.md

@@ -15,6 +15,9 @@ Increment the:
 
 ## [Unreleased]
 
+* [BUG] Elasticsearch: report an asynchronous export's outcome exactly once
+  ([#4501](https://github.com/open-telemetry/opentelemetry-cpp/pull/4501))
+
 * [CONFIGURATION] Apply general `attribute_limits` per individual limit field.
   If a model-specific limit is set it is used, otherwise the matching general
   limit, otherwise the model-specific default. Limit fields on

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -267,74 +267,119 @@ class AsyncResponseHandler : public http_client::EventHandler
   /**
    * Cleans up the session in the destructor.
    */
-  ~AsyncResponseHandler() override { session_->FinishSession(); }
+  ~AsyncResponseHandler() override
+  {
+    // An outcome is owed even here, or a waiter is left on a session that cannot finish.
+    // Reported before FinishSession(), which can block.
+    CompleteOnce(sdk::common::ExportResult::kFailure);
+    session_->FinishSession();
+  }
+
+  /**
+   * Report the outcome of this export, at most once. The HTTP client can deliver both a response
+   * and a terminal session event for one request, and the exporter counts one finished session
+   * per export, so only the first outcome is reported.
+   * @return whether this call is the one that reported.
+   */
+  bool CompleteOnce(sdk::common::ExportResult result) noexcept
+  {
+    bool expected = false;
+    if (!completed_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
+    {
+      return false;
+    }
+    result_callback_(result);
+    return true;
+  }
 
   /**
    * Automatically called when the response is received
    */
   void OnResponse(http_client::Response &response) noexcept override
   {
 
-    // Store the body of the response
-    body_ = std::string(response.GetBody().begin(), response.GetBody().end());
+    const std::string body(response.GetBody().begin(), response.GetBody().end());
+    const bool written = body.find("\"failed\" : 0") != std::string::npos;
+
+    // Reported before anything is logged. CompleteOnce() retires the session and wakes
+    // ForceFlush() before it returns, and the log handler is replaceable, so one that calls
+    // ForceFlush() would otherwise wait for the session this call has not let go of. A response
+    // that loses the exchange says nothing either, since the outcome it would describe is not the
+    // one the caller was given.
+    if (!CompleteOnce(written ? sdk::common::ExportResult::kSuccess
+                              : sdk::common::ExportResult::kFailure))
+    {
+      return;
+    }
+
     if (console_debug_)
     {
       OTEL_INTERNAL_LOG_DEBUG(
-          "[ES Log Exporter] Got response from Elasticsearch,  response body: " << body_);
+          "[ES Log Exporter] Got response from Elasticsearch,  response body: " << body);
     }
-    if (body_.find("\"failed\" : 0") == std::string::npos)
+    if (!written)
     {
       OTEL_INTERNAL_LOG_ERROR(
           "[ES Log Exporter] Logs were not written to Elasticsearch correctly, response body: "
-          << body_);
-      result_callback_(sdk::common::ExportResult::kFailure);
-    }
-    else
-    {
-      result_callback_(sdk::common::ExportResult::kSuccess);
+          << body);
     }
   }
 
   // Callback method when an http event occurs
   void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override
   {
-    bool need_stop = false;
+    // No default label, so -Wswitch reports a state added upstream rather than leaving it
+    // uncounted.
+    const char *failure = nullptr;
     switch (state)
     {
+      // On the way to an outcome, so nothing to report and, in particular, nothing to log: the
+      // session is still registered, and a replaceable log handler that flushed from here would
+      // wait on the export whose call stack it is standing in.
+      case http_client::SessionState::Created:
+      case http_client::SessionState::Connecting:
+      case http_client::SessionState::Connected:
+      case http_client::SessionState::Sending:
+      // The body arrives through OnResponse(), which is what reports the outcome.
+      case http_client::SessionState::Response:
+        break;
       case http_client::SessionState::CreateFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Create request to elasticsearch failed");
-        need_stop = true;
+        failure = "[ES Log Exporter] Create request to elasticsearch failed";
+        break;
+      case http_client::SessionState::Destroyed:
+        failure = "[ES Log Exporter] Session to elasticsearch destroyed before a response";
         break;
       case http_client::SessionState::ConnectFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Connection to elasticsearch failed");
-        need_stop = true;
+        failure = "[ES Log Exporter] Connection to elasticsearch failed";
         break;
       case http_client::SessionState::SendFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request failed to be sent to elasticsearch");
-        need_stop = true;
+        failure = "[ES Log Exporter] Request failed to be sent to elasticsearch";
         break;
       case http_client::SessionState::SSLHandshakeFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] SSL handshake to elasticsearch failed");
-        need_stop = true;
+        failure = "[ES Log Exporter] SSL handshake to elasticsearch failed";
         break;
       case http_client::SessionState::TimedOut:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request to elasticsearch timed out");
-        need_stop = true;
+        failure = "[ES Log Exporter] Request to elasticsearch timed out";
         break;
       case http_client::SessionState::NetworkError:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Network error to elasticsearch");
-        need_stop = true;
+        failure = "[ES Log Exporter] Network error to elasticsearch";
         break;
-      case http_client::SessionState::Cancelled:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request to elasticsearch cancelled");
-        need_stop = true;
+      case http_client::SessionState::ReadError:
+        failure = "[ES Log Exporter] Read error";
+        break;
+      case http_client::SessionState::WriteError:
+        failure = "[ES Log Exporter] Write error";
         break;
-      default:
+      case http_client::SessionState::Cancelled:
+        failure = "[ES Log Exporter] Request to elasticsearch cancelled";
         break;
     }
-    if (need_stop)
+
+    // Logged only when this event is the outcome. These can arrive after a response, and an
+    // error line there would describe a failure the caller was never told about.
+    if (failure != nullptr && CompleteOnce(sdk::common::ExportResult::kFailure))
     {
-      result_callback_(sdk::common::ExportResult::kFailure);
+      OTEL_INTERNAL_LOG_ERROR(failure);
     }
   }
 
@@ -344,8 +389,8 @@ class AsyncResponseHandler : public http_client::EventHandler
   // Callback to call to on receiving events
   std::function<bool(opentelemetry::sdk::common::ExportResult)> result_callback_;
 
-  // A string to store the response body
-  std::string body_ = "";
+  // Whether the outcome has already been reported
+  std::atomic<bool> completed_{false};
 
   // Whether to print the results from the callback
   bool console_debug_ = false;
@@ -446,12 +491,12 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
         {
           OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export "
                                                << span_count
-                                               << " trace span(s) error: " << static_cast<int>(result));
+                                               << " log record(s) error: " << static_cast<int>(result));
         }
         else
         {
           OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count
-                                                                           << " trace span(s) success");
+                                                                           << " log record(s) success");
         }
 
         synchronization_data->finished_session_counter_.fetch_add(1, std::memory_order_release);