Require one bulk item result per exported record

An errors:false body was trusted even when it carried no item results, so
a truncated or unrelated response such as {"errors":false} or an empty
"items" array reported a dropped batch as a success. The check now
requires "items" to be an array whose size matches the number of records
exported: both the sync and async paths pass that count, and the async
handler stores it alongside the response.

Also move the non-2xx status check inside the exception guard so building
its failure reason cannot terminate a noexcept caller on allocation
failure, drop the ?pretty query now that the parser no longer depends on
whitespace, and log a synchronous non-2xx response once from Export()
rather than a second time with the full body from the response handler.

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

Author: thc1006

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

@@ -3,6 +3,7 @@
 
 #pragma once
 
+#include <cstddef>
 #include <nlohmann/json.hpp>
 #include <string>
 
@@ -33,24 +34,33 @@ namespace detail
  * 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 (status_code < 200 || status_code > 299)
-  {
-    failure_reason = "unexpected HTTP status " + std::to_string(status_code);
-    return false;
-  }
 #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);
@@ -67,33 +77,45 @@ 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.
-    const auto items = parsed.find("items");
-    if (items != parsed.end() && items->is_array())
+    for (const auto &item : *items)
     {
-      for (const auto &item : *items)
+      if (!item.is_object())
+      {
+        continue;
+      }
+      for (auto operation = item.begin(); operation != item.end(); ++operation)
       {
-        if (!item.is_object())
+        if (!operation->is_object())
         {
           continue;
         }
-        for (auto operation = item.begin(); operation != item.end(); ++operation)
+        const auto error = operation->find("error");
+        if (error != operation->end())
         {
-          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 = "at least one item failed, first error: " + error->dump();
+          return false;
         }
       }
     }

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -75,35 +75,22 @@ 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.
+      // 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
@@ -234,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}
   {}
 
@@ -264,7 +253,8 @@ 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_, 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, "
                               << failure_reason << ", response body: " << body_);
@@ -328,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;
 };
@@ -386,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");
 
@@ -438,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
@@ -467,7 +460,8 @@ 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, 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, "
                             << failure_reason << ", response body: " << responseBody);

exporters/elasticsearch/test/es_log_record_exporter_test.cc

@@ -164,60 +164,91 @@ constexpr const char *kOneItemRejected =
 TEST(ElasticsearchBulkResponseTests, ReportsSuccessWhenErrorsIsFalse)
 {
   std::string reason;
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kPrettySuccess, reason));
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kCompactSuccess, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kPrettySuccess, 1, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kCompactSuccess, 1, reason));
 }
 
 TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenAnyItemWasRejected)
 {
   std::string reason;
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, kOneItemRejected, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, kOneItemRejected, 2, 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(logs_exporter::detail::IsBulkResponseSuccessful(200, "not json at all", 1, reason));
   EXPECT_FALSE(reason.empty());
 
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "", reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "", 1, 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(logs_exporter::detail::IsBulkResponseSuccessful(200, "[1,2,3]", 1, reason));
   EXPECT_FALSE(reason.empty());
 }
 
 TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNotBoolean)
 {
   std::string reason;
   EXPECT_FALSE(
-      logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"took":30,"items":[]})", reason));
+      logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"took":30,"items":[]})", 0, reason));
   EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":"false","items":[]})", reason));
+      200, R"({"errors":"false","items":[]})", 0, 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, 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));
+  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));
 }
 
 // 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));
+                                                               0, reason));
   EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":true,"items":[null]})", reason));
+      200, R"({"errors":true,"items":[null]})", 1, reason));
   EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
-      200, R"({"errors":true,"items":[{"index":42}]})", reason));
+      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));
 }