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

CHANGELOG.md

@@ -22,6 +22,9 @@ Increment the:
 * [CODE HEALTH] Move remaining API test helpers into anonymous namespaces
   [#4301](https://github.com/open-telemetry/opentelemetry-cpp/pull/4301)
 
+* [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,118 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#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. A successful (2xx) HTTP response whose top-level "errors" is false is treated as
+ * batch success, which is the batch-level signal the exporter needs.
+ *
+ * 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.
+ *
+ * @param status_code the HTTP status code of the response
+ * @param body the raw response body
+ * @param failure_reason set to a best-effort explanation when the function returns false; an
+ *        allocation failure on an error path may leave it empty
+ * @return true when the status is 2xx and the response reports the batch as written
+ */
+inline bool IsBulkResponseSuccessful(int status_code,
+                                     const std::string &body,
+                                     std::string &failure_reason) noexcept
+{
+  failure_reason.clear();
+#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;
+    }
+
+    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;
+    }
+
+    if (!errors->get<bool>())
+    {
+      return true;
+    }
+
+    // errors is true: report the first item error rather than only saying that something failed.
+    const auto items = parsed.find("items");
+    if (items != parsed.end() && items->is_array())
+    {
+      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;
 };
@@ -245,11 +250,11 @@ 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_, 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
@@ -368,7 +373,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");
 
@@ -448,11 +453,11 @@ 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, 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;
   }

exporters/elasticsearch/test/es_log_record_exporter_test.cc

@@ -3,6 +3,7 @@
 
 #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h"
 #include "opentelemetry/common/timestamp.h"
+#include "opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h"
 #include "opentelemetry/exporters/elasticsearch/es_log_recordable.h"
 #include "opentelemetry/logs/severity.h"
 #include "opentelemetry/nostd/span.h"
@@ -142,3 +143,111 @@ TEST(ElasticsearchLogRecordableTests, BasicTests)
 
   EXPECT_EQ(actual, expected);
 }
+
+// The bulk API reports the outcome for the batch in the top level "errors" field. These cover
+// the two cases the previous substring test got wrong: a batch that had a rejected item was
+// reported as a success, and a successful batch from a server that does not pretty-print was
+// reported as a failure.
+namespace
+{
+constexpr const char *kPrettySuccess =
+    R"({"took":30,"errors":false,"items":[{"index":{"_shards":{"total":2,"successful":1,)"
+    R"("failed" : 0}}}]})";
+constexpr const char *kCompactSuccess =
+    R"({"took":30,"errors":false,"items":[{"index":{"_shards":{"total":2,"successful":1,)"
+    R"("failed":0}}}]})";
+constexpr const char *kOneItemRejected =
+    R"({"took":30,"errors":true,"items":[{"index":{"_shards":{"failed" : 0}}},)"
+    R"({"index":{"error":{"type":"mapper_parsing_exception","reason":"bad field"}}}]})";
+}  // namespace
+
+TEST(ElasticsearchBulkResponseTests, ReportsSuccessWhenErrorsIsFalse)
+{
+  std::string reason;
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kPrettySuccess, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kCompactSuccess, reason));
+}
+
+// A successful check must not leave a stale failure reason from a previous call behind.
+TEST(ElasticsearchBulkResponseTests, ClearsFailureReasonOnSuccess)
+{
+  std::string reason = "stale";
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kCompactSuccess, reason));
+  EXPECT_TRUE(reason.empty());
+}
+
+TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenAnyItemWasRejected)
+{
+  std::string reason;
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, kOneItemRejected, reason));
+  EXPECT_NE(reason.find("mapper_parsing_exception"), std::string::npos);
+}
+
+TEST(ElasticsearchBulkResponseTests, ReportsFailureOnUnusableBody)
+{
+  std::string reason;
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "not json at all", reason));
+  EXPECT_FALSE(reason.empty());
+
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "", reason));
+  EXPECT_FALSE(reason.empty());
+
+  // An array rather than the expected object.
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "[1,2,3]", reason));
+  EXPECT_FALSE(reason.empty());
+}
+
+TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNotBoolean)
+{
+  std::string reason;
+  EXPECT_FALSE(
+      logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"took":30,"items":[]})", reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
+      200, R"({"errors":"false","items":[]})", reason));
+}
+
+// A non-2xx status is a failure even when the body reports errors:false. This is the invariant
+// the substring check and the body-only check both missed, on both the sync and async paths.
+TEST(ElasticsearchBulkResponseTests, ReportsFailureOnNon2xxStatus)
+{
+  std::string reason;
+  const char *ok_body = R"({"errors":false,"items":[]})";
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(500, ok_body, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(429, ok_body, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(400, ok_body, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, ok_body, reason));
+}
+
+// errors:true with no extractable item error still fails, with the generic reason.
+TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsTrueWithoutItemError)
+{
+  std::string reason;
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"errors":true,"items":[]})",
+                                                               reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
+      200, R"({"errors":true,"items":[null]})", reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
+      200, R"({"errors":true,"items":[{"index":42}]})", reason));
+}
+
+// The 2xx range is the success band; the codes just outside it are failures. This pins the boundary
+// so a later change to the status check cannot silently widen or narrow it.
+TEST(ElasticsearchBulkResponseTests, TreatsThe2xxRangeAsTheSuccessBand)
+{
+  std::string reason;
+  const char *ok_body = R"({"errors":false,"items":[]})";
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(199, ok_body, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, ok_body, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(299, ok_body, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(300, ok_body, reason));
+}
+
+// Batch success is decided by the top-level "errors" flag alone. "items" is only read to extract a
+// diagnostic reason on failure, so a response without "items" (for example one filtered with
+// filter_path) is still a success when "errors" is false. This locks out re-introducing an
+// items-count success check.
+TEST(ElasticsearchBulkResponseTests, DoesNotRequireItemsForSuccess)
+{
+  std::string reason;
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"errors":false})", reason));
+}