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

CHANGELOG.md

@@ -150,6 +150,14 @@ 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 that names its target index
+  and says the operation applied
+  [#4297](https://github.com/open-telemetry/opentelemetry-cpp/pull/4297)
+* [BUG] Stop reading an Elasticsearch bulk response that only says the request
+  was accepted as a batch that was written
+  [#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)
@@ -55,4 +61,10 @@ if(BUILD_TESTING)
     TARGET es_log_record_exporter_test
     TEST_PREFIX exporter.
     TEST_LIST es_log_record_exporter_test)
+
+  # The synchronous wiring cases wait on a completion, and a regression that
+  # stops delivering one hangs the job rather than reporting. A bound turns that
+  # back into a red test. Not the asynchronous ones: their fake calls back from
+  # inside SendRequest() and Export() returns without waiting.
+  set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 120)
 endif() # BUILD_TESTING

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

@@ -0,0 +1,236 @@
+// 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
+{
+
+/**
+ * What the bulk request itself has to answer with. Elasticsearch answers a bulk request with 200,
+ * and the rest of the 2xx band does not say the batch was written. 202 is the one that matters:
+ * it says the request was accepted and the processing is not finished, and may never be. Reading
+ * that as success is not a conservative reading of a vague answer, it is the opposite of what the
+ * answer says, and a success here lets the caller drop the records it just handed over.
+ */
+inline bool IsSuccessfulBulkStatus(int status_code) noexcept
+{
+  return 200 == status_code;
+}
+
+/**
+ * What an acknowledged index operation has to carry. Elasticsearch answers one with 201 when it
+ * created the document and 200 when it replaced an existing one.
+ *
+ * Two overloads rather than one signed parameter, because an operation status arrives through
+ * nlohmann::json and is compared in the type it was parsed as: is_number_integer() is true for
+ * unsigned as well, and 2^32 + 200 narrows back to 200 on a 32 bit int.
+ */
+inline bool IsAppliedStatus(nlohmann::json::number_unsigned_t value) noexcept
+{
+  return 200U == value || 201U == value;
+}
+
+inline bool IsAppliedStatus(nlohmann::json::number_integer_t value) noexcept
+{
+  return 200 == value || 201 == value;
+}
+
+/**
+ * Whether an acknowledged operation status is one that applied the operation.
+ *
+ * The type check is here rather than left to the caller, although the one caller does it too.
+ * This is noexcept, and get() on a value that is not a number throws, so a caller that forgot
+ * would not get a wrong answer, it would get std::terminate. A float does not throw: it
+ * truncates, so 200.5 would answer for 200.
+ */
+inline bool IsAcknowledgedStatus(const nlohmann::json &status) noexcept
+{
+  if (!status.is_number_integer())
+  {
+    return false;
+  }
+
+  if (status.is_number_unsigned())
+  {
+    return IsAppliedStatus(status.get<nlohmann::json::number_unsigned_t>());
+  }
+
+  return IsAppliedStatus(status.get<nlohmann::json::number_integer_t>());
+}
+
+/**
+ * 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.
+ *
+ * This reads the fields the bulk response documents to decide an outcome. It is not a check
+ * against a responder that is trying to be believed: a repeated key, for one, is folded before
+ * this sees the document, so a body carrying both "errors": true and "errors": false arrives as
+ * whichever one the parser kept, and nothing here can tell that the other was ever sent.
+ *
+ * @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 200, "errors" is false, and "items" holds exactly
+ *         expected_items index results that each name a target index, acknowledge a status the
+ *         operation applied under, and carry no error
+ */
+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.
+    // Not the band an operation status is held to. This one answers for the request, and an
+    // operation that applied answers 201 for a document it created, which the request itself
+    // never says.
+    if (!IsSuccessfulBulkStatus(status_code))
+    {
+      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.
+    // An operation that was not applied says so twice, through its status and through an "error"
+    // member. Both are read here so that neither reading depends on the "errors" flag, which a
+    // response that contradicts itself also controls.
+    const nlohmann::json *rejected   = nullptr;
+    const nlohmann::json *item_error = 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;
+      }
+
+      // Elasticsearch names the index it wrote to in every index result, whether the operation
+      // applied or not. Presence and type only: the name it reports is the index the write
+      // resolved to, which an alias or a date math index makes different from the one that was
+      // submitted, so it is not something to compare against.
+      const auto target = operation->find("_index");
+      if (target == operation->end() || !target->is_string())
+      {
+        failure_reason = "the response acknowledges an index operation with no target index";
+        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);
+      }
+
+      // A null holds no cause, and a serialiser that writes absent optionals as null is saying
+      // the operation applied, which is what the flag says too. Only a cause contradicts it.
+      const auto error = operation->find("error");
+      if (item_error == nullptr && error != operation->end() && !error->is_null())
+      {
+        item_error = &(*error);
+      }
+    }
+
+    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;
+      }
+      if (item_error != nullptr)
+      {
+        failure_reason =
+            "the response reports no errors but an item carries error " + item_error->dump();
+        return false;
+      }
+      return true;
+    }
+
+    // Name the first item error rather than only saying that something failed.
+    if (item_error != nullptr)
+    {
+      failure_reason = "at least one item failed, first error: " + item_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