Require an acknowledgement for every submitted operation

The exporter posts an unfiltered /_bulk with one index operation per
record, and Elasticsearch answers those with one entry in items per
operation. The check accepted any 2xx body carrying errors:false, so
{"errors":false\} alone was reported as a successful write of every
record in the batch.

An exporter that reports success without a write acknowledgement gives
the caller no reason to retry, which is the more dangerous direction of
the two. A misconfigured proxy, a non-Elasticsearch endpoint answering
with coincidentally shaped JSON, or a server returning an incomplete
envelope all reached the same result.

An earlier revision of this branch dropped a count check as half baked
schema validation, on the grounds that a filter_path response need not
carry items. That reasoning does not apply to this exporter: it never
sends filter_path. Confirming the response answers as many operations as
were submitted is not the same as validating each item's contents.

Author: thc1006

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

@@ -42,6 +42,7 @@ namespace detail
  */
 inline bool IsBulkResponseSuccessful(int status_code,
                                      const std::string &body,
+                                     std::size_t expected_items,
                                      std::string &failure_reason) noexcept
 {
   failure_reason.clear();
@@ -70,14 +71,30 @@ inline bool IsBulkResponseSuccessful(int status_code,
       return false;
     }
 
+    // The exporter posts an unfiltered /_bulk, one index operation per record, and Elasticsearch
+    // answers those with one entry in "items" per operation. A body without that many entries is
+    // not an answer to the request that was sent, so it is not evidence that the records were
+    // written, whatever "errors" says.
+    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;
+    }
+
     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)
       {

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -256,9 +256,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 submitted_operations,
       bool console_debug = false)
       : session_{std::move(session)},
         result_callback_{std::move(result_callback)},
+        submitted_operations_{submitted_operations},
         console_debug_{console_debug}
   {}
 
@@ -286,7 +288,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_, submitted_operations_,
+                                          failure_reason))
     {
       OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, "
                               << failure_reason << ", response body: " << body_);
@@ -347,6 +350,9 @@ class AsyncResponseHandler : public http_client::EventHandler
   // Callback to call to on receiving events
   std::function<bool(opentelemetry::sdk::common::ExportResult)> result_callback_;
 
+  // How many operations this request submitted, which is how many the response has to answer
+  std::size_t submitted_operations_ = 0;
+
   // A string to store the response body
   std::string body_ = "";
 
@@ -460,7 +466,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
@@ -489,7 +495,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

@@ -167,46 +167,46 @@ 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));
 }
 
 // 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(logs_exporter::detail::IsBulkResponseSuccessful(200, kCompactSuccess, 1, reason));
   EXPECT_TRUE(reason.empty());
 }
 
 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":[]})", 1, 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
@@ -215,22 +215,22 @@ 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));
+  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));
 }
 
 // The 2xx range is the success band; the codes just outside it are failures. This pins the boundary
@@ -239,20 +239,32 @@ 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));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(199, ok_body, 0, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, ok_body, 0, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(299, ok_body, 0, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(300, ok_body, 0, 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)
+// The exporter posts an unfiltered /_bulk, so Elasticsearch answers every submitted operation in
+// "items". A body that does not is not an answer to that request, and treating it as success would
+// hand the caller a write acknowledgement nobody made.
+TEST(ElasticsearchBulkResponseTests, RequiresAnAcknowledgementForEverySubmittedOperation)
 {
   std::string reason;
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"errors":false})", reason));
+  constexpr const char *kNoItems   = R"({"errors":false})";
+  constexpr const char *kNullItems = R"({"errors":false,"items":null})";
+  constexpr const char *kOneItemBody =
+      R"({"errors":false,"items":[{"index":{"_shards":{"failed":0}}}]})";
+  constexpr const char *kTwoItemBody = R"({"errors":false,"items":[{"index":{}},{"index":{}}]})";
+
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, kNoItems, 1, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, kNullItems, 1, reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, kOneItemBody, 2, reason))
+      << "too few acknowledgements";
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, kTwoItemBody, 1, reason))
+      << "more acknowledgements than operations";
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kOneItemBody, 1, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kTwoItemBody, 2, reason));
 }
 // ---------------------------------------------------------------------------
 // The response travelling from the HTTP callback to the verdict.