[BUG] Decide Elasticsearch bulk export success from HTTP status and the errors flag

CHANGELOG.md

@@ -15,6 +15,9 @@ Increment the:
 
 ## [Unreleased]
 
+* [BUG] Parse the Elasticsearch bulk response instead of matching a substring
+  [#4297](https://github.com/open-telemetry/opentelemetry-cpp/pull/4297)
+
 * [CODE HEALTH] Move metrics storage test fixtures into anonymous namespace
   [#4286](https://github.com/open-telemetry/opentelemetry-cpp/pull/4286)
 

exporters/elasticsearch/BUILD

@@ -13,6 +13,7 @@ cc_library(
         "src/es_log_recordable.cc",
     ],
     hdrs = [
+        "include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h",
         "include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h",
         "include/opentelemetry/exporters/elasticsearch/es_log_recordable.h",
     ],

exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h

@@ -0,0 +1,137 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <cstddef>
+#include <nlohmann/json.hpp>
+#include <string>
+
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace exporter
+{
+namespace logs
+{
+namespace detail
+{
+
+/**
+ * Decide whether an Elasticsearch bulk response reports the whole batch as written.
+ *
+ * The bulk API reports the outcome for the batch in the top level "errors" field. A "failed"
+ * count belongs to the per-shard information of a single item and says nothing about the other
+ * items, so matching it as a substring both misses real failures and depends on how the
+ * response happens to be formatted.
+ *
+ * Callers include noexcept response handlers, so nothing may escape from here. Anything that
+ * stops the body being inspected counts as a failed export. With exceptions disabled there is
+ * nothing to catch, since an allocation failure terminates instead.
+ *
+ * A non-2xx HTTP status is a failure regardless of the body: the top-level "errors" flag only
+ * describes item-level outcomes and cannot override a transport or application error, so both
+ * the synchronous and asynchronous paths must pass the status here rather than inspecting the
+ * body alone.
+ *
+ * A compliant bulk response also carries one entry in "items" per submitted operation. A response
+ * that omits "items" or returns a different count cannot confirm the batch was written, so it is
+ * a failure too: otherwise an "errors":false body with no items would report a dropped batch as a
+ * success. Callers pass the number of records they exported so the counts can be compared.
+ *
+ * @param status_code the HTTP status code of the response
+ * @param body the raw response body
+ * @param expected_item_count the number of records submitted in the bulk request
+ * @param failure_reason set to a short explanation when the function returns false
+ * @return true when the status is 2xx and the response reports every item as written
+ */
+inline bool IsBulkResponseSuccessful(int status_code,
+                                     const std::string &body,
+                                     std::size_t expected_item_count,
+                                     std::string &failure_reason) noexcept
+{
+#if OPENTELEMETRY_HAVE_EXCEPTIONS
+  try
+  {
+#endif
+    // Inside the try so that even the string building below cannot escape a noexcept caller.
+    if (status_code < 200 || status_code > 299)
+    {
+      failure_reason = "unexpected HTTP status " + std::to_string(status_code);
+      return false;
+    }
+
+    failure_reason = "the response body could not be inspected";
+
+    const nlohmann::json parsed = nlohmann::json::parse(body, nullptr, false);
+    if (parsed.is_discarded() || !parsed.is_object())
+    {
+      failure_reason = "the response body is not a JSON object";
+      return false;
+    }
+
+    const auto errors = parsed.find("errors");
+    if (errors == parsed.end() || !errors->is_boolean())
+    {
+      failure_reason = "the response body has no boolean \"errors\" field";
+      return false;
+    }
+
+    // Require one item result per exported record before trusting the outcome flag, so a
+    // truncated or unrelated body cannot be read as a whole-batch success.
+    const auto items = parsed.find("items");
+    if (items == parsed.end() || !items->is_array())
+    {
+      failure_reason = "the response body has no array \"items\" field";
+      return false;
+    }
+    if (items->size() != expected_item_count)
+    {
+      failure_reason = "the response reported " + std::to_string(items->size()) +
+                       " item result(s) for " + std::to_string(expected_item_count) +
+                       " exported record(s)";
+      return false;
+    }
+
+    if (!errors->get<bool>())
+    {
+      return true;
+    }
+
+    // Report the first item error rather than only saying that something failed.
+    for (const auto &item : *items)
+    {
+      if (!item.is_object())
+      {
+        continue;
+      }
+      for (auto operation = item.begin(); operation != item.end(); ++operation)
+      {
+        if (!operation->is_object())
+        {
+          continue;
+        }
+        const auto error = operation->find("error");
+        if (error != operation->end())
+        {
+          failure_reason = "at least one item failed, first error: " + error->dump();
+          return false;
+        }
+      }
+    }
+
+    failure_reason = "the response reports errors";
+    return false;
+#if OPENTELEMETRY_HAVE_EXCEPTIONS
+  }
+  catch (...)
+  {
+    return false;
+  }
+#endif
+}
+
+}  // namespace detail
+}  // namespace logs
+}  // namespace exporter
+OPENTELEMETRY_END_NAMESPACE

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -14,6 +14,7 @@
 #include <utility>
 #include <vector>
 
+#include "opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h"
 #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h"
 #include "opentelemetry/exporters/elasticsearch/es_log_recordable.h"
 #include "opentelemetry/ext/http/client/detail/default_factory.h"
@@ -41,6 +42,7 @@ namespace exporter
 {
 namespace logs
 {
+
 /**
  * This class handles the response message from the Elasticsearch request
  */
@@ -73,33 +75,24 @@ class ResponseHandler : public http_client::EventHandler
    */
   void OnResponse(http_client::Response &response) noexcept override
   {
-    std::string log_message;
-
     // Lock the private members so they can't be read while being modified
     {
       std::unique_lock<std::mutex> lk(mutex_);
 
       // Store the body of the request
       body_ = std::string(response.GetBody().begin(), response.GetBody().end());
 
-      if (!(response.GetStatusCode() >= 200 && response.GetStatusCode() <= 299))
-      {
-        log_message = BuildResponseLogMessage(response, body_);
-
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Export failed, " << log_message);
-      }
-
       if (console_debug_)
       {
-        if (log_message.empty())
-        {
-          log_message = BuildResponseLogMessage(response, body_);
-        }
-
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Got response from Elasticsearch, "
-                                << log_message);
+                                << BuildResponseLogMessage(response, body_));
       }
 
+      // Keep the status so Export() can fold it into the ExportResult; storing the body alone
+      // let a non-2xx response be reported as a success. Export() is the single place that logs
+      // the failure, so a non-2xx response is not logged a second time here with the full body.
+      status_code_ = response.GetStatusCode();
+
       // Set the response_received_ flag to true and notify any threads waiting on this result
       response_received_ = true;
     }
@@ -127,6 +120,15 @@ class ResponseHandler : public http_client::EventHandler
     return body_;
   }
 
+  /**
+   * Returns the HTTP status code of the response
+   */
+  int GetStatusCode()
+  {
+    std::unique_lock<std::mutex> lk(mutex_);
+    return status_code_;
+  }
+
   // Callback method when an http event occurs
   void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override
   {
@@ -199,6 +201,9 @@ class ResponseHandler : public http_client::EventHandler
   // A string to store the response body
   std::string body_ = "";
 
+  // The HTTP status code of the response
+  int status_code_ = 0;
+
   // Whether to print the results from the callback
   bool console_debug_ = false;
 };
@@ -216,9 +221,11 @@ class AsyncResponseHandler : public http_client::EventHandler
   AsyncResponseHandler(
       std::shared_ptr<ext::http::client::Session> session,
       std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback,
+      std::size_t expected_item_count,
       bool console_debug = false)
       : session_{std::move(session)},
         result_callback_{std::move(result_callback)},
+        expected_item_count_{expected_item_count},
         console_debug_{console_debug}
   {}
 
@@ -245,11 +252,12 @@ class AsyncResponseHandler : public http_client::EventHandler
       OTEL_INTERNAL_LOG_DEBUG(
           "[ES Log Exporter] Got response from Elasticsearch,  response body: " << body_);
     }
-    if (body_.find("\"failed\" : 0") == std::string::npos)
+    std::string failure_reason;
+    if (!detail::IsBulkResponseSuccessful(response.GetStatusCode(), body_, expected_item_count_,
+                                          failure_reason))
     {
-      OTEL_INTERNAL_LOG_ERROR(
-          "[ES Log Exporter] Logs were not written to Elasticsearch correctly, response body: "
-          << body_);
+      OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, "
+                              << failure_reason << ", response body: " << body_);
       result_callback_(sdk::common::ExportResult::kFailure);
     }
     else
@@ -310,6 +318,9 @@ class AsyncResponseHandler : public http_client::EventHandler
   // A string to store the response body
   std::string body_ = "";
 
+  // The number of records submitted, used to check the response reports a result for each
+  std::size_t expected_item_count_ = 0;
+
   // Whether to print the results from the callback
   bool console_debug_ = false;
 };
@@ -368,7 +379,7 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
   auto request = session->CreateRequest();
 
   // Populate the request with headers and methods
-  request->SetUri(options_.index_ + "/_bulk?pretty");
+  request->SetUri(options_.index_ + "/_bulk");
   request->SetMethod(http_client::Method::Post);
   request->AddHeader("Content-Type", "application/json");
 
@@ -420,7 +431,7 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
         synchronization_data->force_flush_cv.notify_all();
         return true;
       },
-      options_.console_debug_);
+      span_count, options_.console_debug_);
   session->SendRequest(handler);
   return sdk::common::ExportResult::kSuccess;
 #else
@@ -448,11 +459,12 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
 
   // Parse the response output to determine if Elasticsearch consumed it correctly
   std::string responseBody = handler->GetResponseBody();
-  if (responseBody.find("\"failed\" : 0") == std::string::npos)
+  std::string failure_reason;
+  if (!detail::IsBulkResponseSuccessful(handler->GetStatusCode(), responseBody, records.size(),
+                                        failure_reason))
   {
-    OTEL_INTERNAL_LOG_ERROR(
-        "[ES Log Exporter] Logs were not written to Elasticsearch correctly, response body: "
-        << responseBody);
+    OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, "
+                            << failure_reason << ", response body: " << responseBody);
     // TODO: Retry logic
     return sdk::common::ExportResult::kFailure;
   }