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

The exporter decided a bulk export had succeeded by searching the response body
for the substring "failed" : 0. That is a shard counter belonging to one item,
not a verdict on the batch, so a body that never contained it read as a failure
and a body that contained it anywhere read as a success.

Decide from the documented contract instead: a 2xx HTTP status, a top level
"errors" flag that is false, and one acknowledged operation result per record
submitted. Each entry of "items" has to be the result of the index operation the
exporter actually sent, carrying an integer status in the 2xx band, so a well
formed body that acknowledges nothing cannot pass. When "errors" is true the
first rejected item names the status in the log.

The status is compared in the type it was parsed as. Reading it through an int
first is not safe: the value comes from the server, is_number_integer() holds
for unsigned as well, and 2^32 + 200 narrows back into the 2xx band on a 32 bit
int. Comparing directly also removes the need for a rejected-status sentinel,
which could not tell a real status of 0 apart from "nothing was rejected".

The parser lives in detail/es_bulk_response.h, excluded from both the CMake
install set and the Bazel public headers.

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

Author: thc1006

CHANGELOG.md

@@ -65,6 +65,10 @@ Increment the:
 * [BUG] Wait on a completion state in the Elasticsearch exporter
   [#4298](https://github.com/open-telemetry/opentelemetry-cpp/pull/4298)
 
+* [BUG] Decide Elasticsearch bulk export success from the HTTP status, the
+  errors flag, and one acknowledgement per record
+  [#4297](https://github.com/open-telemetry/opentelemetry-cpp/pull/4297)
+
 * [BUG] Make SocketAddr string parsing safe and reject malformed addresses
   [#4292](https://github.com/open-telemetry/opentelemetry-cpp/pull/4292)
 

exporters/elasticsearch/BUILD

@@ -6,6 +6,24 @@ load("@rules_cc//cc:cc_test.bzl", "cc_test")
 
 package(default_visibility = ["//visibility:public"])
 
+# The bulk response parser is an implementation detail that the CMake package
+# also leaves out. hdrs is a target's public interface, so it is kept in its own
+# target that only this package can depend on, and reached through
+# implementation_deps so the exporter does not re-export it.
+cc_library(
+    name = "es_bulk_response",
+    hdrs = [
+        "include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h",
+    ],
+    strip_include_prefix = "include",
+    tags = ["es"],
+    visibility = ["//exporters/elasticsearch:__pkg__"],
+    deps = [
+        "//api",
+        "@github_nlohmann_json//:json",
+    ],
+)
+
 cc_library(
     name = "es_log_record_exporter",
     srcs = [
@@ -16,6 +34,7 @@ cc_library(
         "include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h",
         "include/opentelemetry/exporters/elasticsearch/es_log_recordable.h",
     ],
+    implementation_deps = [":es_bulk_response"],
     linkopts = select({
         "//bazel:windows": [
             "-DEFAULTLIB:advapi32.lib",
@@ -43,6 +62,7 @@ cc_test(
         "test",
     ],
     deps = [
+        ":es_bulk_response",
         ":es_log_record_exporter",
         "@com_google_googletest//:gtest_main",
         "@curl",

exporters/elasticsearch/CMakeLists.txt

@@ -34,6 +34,12 @@ otel_add_component(
   "*.h"
   PATTERN
   "es_log_recordable.h"
+  EXCLUDE
+  PATTERN
+  "es_bulk_response.h"
+  EXCLUDE
+  PATTERN
+  "detail"
   EXCLUDE)
 
 if(OPENTELEMETRY_INSTALL)

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

@@ -0,0 +1,171 @@
+// 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
+{
+
+/**
+ * Whether an acknowledged operation status is one that applied the operation. Elasticsearch
+ * answers an index operation with 200 or 201, so 2xx is the whole band.
+ *
+ * Compared in the type the number was parsed as. Narrowing to int first is not safe here: the
+ * value comes from the server, is_number_integer() is true for unsigned as well, and 2^32 + 200
+ * narrows back into the band on a 32 bit int.
+ */
+inline bool IsAcknowledgedStatus(const nlohmann::json &status) noexcept
+{
+  if (status.is_number_unsigned())
+  {
+    const auto value = status.get<nlohmann::json::number_unsigned_t>();
+    return value >= 200U && value <= 299U;
+  }
+
+  const auto value = status.get<nlohmann::json::number_integer_t>();
+  return value >= 200 && value <= 299;
+}
+
+/**
+ * Decide whether an Elasticsearch bulk response reports the whole batch as written.
+ *
+ * Callers include noexcept response handlers, so nothing may escape. Anything that stops the body
+ * being inspected counts as a failed export.
+ *
+ * @param status_code the response status, which the caller passes rather than this reading the
+ *        body alone: "errors" describes item outcomes and cannot override a transport error
+ * @param body the raw response body
+ * @param expected_items the number of index operations the request submitted
+ * @param failure_reason a best-effort explanation when this returns false
+ * @return true only when the status is 2xx, "errors" is false, and "items" holds exactly
+ *         expected_items index results that each acknowledge a 2xx status
+ */
+inline bool IsBulkResponseSuccessful(int status_code,
+                                     const std::string &body,
+                                     std::size_t expected_items,
+                                     std::string &failure_reason) noexcept
+{
+  failure_reason.clear();
+#if OPENTELEMETRY_HAVE_EXCEPTIONS
+  try
+  {
+#endif
+    // Inside the try: building a reason allocates.
+    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;
+    }
+
+    // The request is an unfiltered /_bulk, so "items" is always there and holds one entry per
+    // operation. The next three checks decide whether this body answers the request that was sent.
+    const auto items = parsed.find("items");
+    if (items == parsed.end() || !items->is_array())
+    {
+      failure_reason = "the response body has no \"items\" array";
+      return false;
+    }
+
+    if (items->size() != expected_items)
+    {
+      failure_reason = "the response acknowledges " + std::to_string(items->size()) + " of " +
+                       std::to_string(expected_items) + " submitted operations";
+      return false;
+    }
+
+    // Each entry is keyed by the action it answers, and every record goes out as an index
+    // operation. Read before "errors", which a body that is not this answer does not get to decide.
+    const nlohmann::json *rejected = nullptr;
+    for (const auto &item : *items)
+    {
+      if (!item.is_object() || item.size() != 1)
+      {
+        failure_reason = "the response has an \"items\" entry that is not one operation result";
+        return false;
+      }
+
+      const auto operation = item.find("index");
+      if (operation == item.end() || !operation->is_object())
+      {
+        failure_reason = "the response does not acknowledge the submitted index operation";
+        return false;
+      }
+
+      const auto status = operation->find("status");
+      if (status == operation->end() || !status->is_number_integer())
+      {
+        failure_reason = "the response acknowledges an index operation with no status";
+        return false;
+      }
+
+      if (rejected == nullptr && !IsAcknowledgedStatus(*status))
+      {
+        rejected = &(*status);
+      }
+    }
+
+    if (!errors->get<bool>())
+    {
+      // A false flag claims every operation was applied; hold the items to that claim.
+      if (rejected != nullptr)
+      {
+        failure_reason =
+            "the response reports no errors but acknowledges operation status " + rejected->dump();
+        return false;
+      }
+      return true;
+    }
+
+    // Name the first item error rather than only saying that something failed. Every entry is one
+    // index result by now, so its single member is the result to look in.
+    for (const auto &item : *items)
+    {
+      const auto &result = *item.begin();
+      const auto error   = result.find("error");
+      if (error != result.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