Drop the item-count check; trust the top-level errors flag

Review found the added items.size() == records check to be half-baked
schema validation beyond the scope of #4295: it checks array length but
not that each entry is a real operation result, and it would reject a
legitimate filter_path response. A 2xx response whose top-level "errors"
is false is Elasticsearch's authoritative batch-success signal, so the
condition returns to 2xx + JSON object + boolean errors. Also clear
failure_reason on entry so a success cannot leave a stale reason behind,
and document the exception path's reason as best-effort.

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

Author: thc1006

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

@@ -3,7 +3,6 @@
 
 #pragma once
 
-#include <cstddef>
 #include <nlohmann/json.hpp>
 #include <string>
 
@@ -20,36 +19,32 @@ 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.
+ * 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 2xx response whose top-level "errors" is false means every operation was applied,
+ * which is the batch-level success 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.
+ * 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.
+ * 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 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
+ * @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::size_t expected_item_count,
                                      std::string &failure_reason) noexcept
 {
+  failure_reason.clear();
 #if OPENTELEMETRY_HAVE_EXCEPTIONS
   try
   {
@@ -61,8 +56,6 @@ inline bool IsBulkResponseSuccessful(int 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())
     {
@@ -77,45 +70,33 @@ inline bool IsBulkResponseSuccessful(int status_code,
       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)
+    // 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())
     {
-      if (!item.is_object())
-      {
-        continue;
-      }
-      for (auto operation = item.begin(); operation != item.end(); ++operation)
+      for (const auto &item : *items)
       {
-        if (!operation->is_object())
+        if (!item.is_object())
         {
           continue;
         }
-        const auto error = operation->find("error");
-        if (error != operation->end())
+        for (auto operation = item.begin(); operation != item.end(); ++operation)
         {
-          failure_reason = "at least one item failed, first error: " + error->dump();
-          return false;
+          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;
+          }
         }
       }
     }

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -221,11 +221,9 @@ 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}
   {}
 
@@ -253,8 +251,7 @@ class AsyncResponseHandler : public http_client::EventHandler
           "[ES Log Exporter] Got response from Elasticsearch,  response body: " << body_);
     }
     std::string failure_reason;
-    if (!detail::IsBulkResponseSuccessful(response.GetStatusCode(), body_, expected_item_count_,
-                                          failure_reason))
+    if (!detail::IsBulkResponseSuccessful(response.GetStatusCode(), body_, failure_reason))
     {
       OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, "
                               << failure_reason << ", response body: " << body_);
@@ -318,9 +315,6 @@ 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;
 };
@@ -431,7 +425,7 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
         synchronization_data->force_flush_cv.notify_all();
         return true;
       },
-      span_count, options_.console_debug_);
+      options_.console_debug_);
   session->SendRequest(handler);
   return sdk::common::ExportResult::kSuccess;
 #else
@@ -460,8 +454,7 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
   // Parse the response output to determine if Elasticsearch consumed it correctly
   std::string responseBody = handler->GetResponseBody();
   std::string failure_reason;
-  if (!detail::IsBulkResponseSuccessful(handler->GetStatusCode(), responseBody, records.size(),
-                                        failure_reason))
+  if (!detail::IsBulkResponseSuccessful(handler->GetStatusCode(), responseBody, failure_reason))
   {
     OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, "
                             << failure_reason << ", response body: " << responseBody);

exporters/elasticsearch/test/es_log_record_exporter_test.cc

@@ -164,91 +164,68 @@ constexpr const char *kOneItemRejected =
 TEST(ElasticsearchBulkResponseTests, ReportsSuccessWhenErrorsIsFalse)
 {
   std::string reason;
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kPrettySuccess, 1, reason));
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kCompactSuccess, 1, 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, 2, 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", 1, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "not json at all", reason));
   EXPECT_FALSE(reason.empty());
 
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "", 1, reason));
+  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]", 1, reason));
+  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":[]})", 0, reason));
+      logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"took":30,"items":[]})", reason));
   EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":"false","items":[]})", 0, reason));
+      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;
-  // Empty batch, so zero item results are expected; the status is what varies here.
   const char *ok_body = R"({"errors":false,"items":[]})";
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(500, ok_body, 0, reason));
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(429, ok_body, 0, reason));
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(400, ok_body, 0, reason));
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, ok_body, 0, reason));
+  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":[]})",
-                                                               0, reason));
+                                                               reason));
   EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":true,"items":[null]})", 1, reason));
+      200, R"({"errors":true,"items":[null]})", reason));
   EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":true,"items":[{"index":42}]})", 1, reason));
-}
-
-// A 2xx errors:false body must still carry one item result per exported record. A missing or short
-// "items" array cannot confirm the batch was written, so a dropped or truncated response that
-// happens to be valid JSON is a failure rather than a silent success.
-TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenItemCountDoesNotMatch)
-{
-  std::string reason;
-
-  // errors:false but no "items" field at all.
-  EXPECT_FALSE(
-      logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"errors":false})", 1, reason));
-  EXPECT_NE(reason.find("items"), std::string::npos);
-
-  // errors:false with an empty "items" array for a non-empty batch.
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":false,"items":[]})", 1, reason));
-
-  // Fewer item results than records exported.
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":false,"items":[{"index":{"status":201}}]})", 2, reason));
-
-  // One result per record is a success.
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":false,"items":[{"index":{"status":201}},{"index":{"status":201}}]})", 2,
-      reason));
-
-  // An empty batch with an empty "items" array is a success.
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"errors":false,"items":[]})",
-                                                              0, reason));
+      200, R"({"errors":true,"items":[{"index":42}]})", reason));
 }