Fold the HTTP status into the bulk export result on both paths

Review of open-telemetry#4297 found the synchronous path reported a non-2xx response
as a success: ResponseHandler::OnResponse only logged the status and
unconditionally set response_received_, waitForResponse returned true,
and Export then checked the body alone, so HTTP 500 with a body of
{"errors":false} became kSuccess. My description claiming the sync path
already checked the status was wrong: it observed the status but never
let it affect the ExportResult.

IsBulkResponseSuccessful now takes the status code and treats a non-2xx
as a failure before looking at the body. The top-level "errors" flag is
item-level and cannot override a transport or application error. Both
paths call it: the async handler drops its duplicated status check, and
the sync handler stores the status so Export can pass it in.

Tests assert the invariant directly, including HTTP 500 with
{"errors":false} on the sync-shaped path, plus the errors:true generic
reason and the missing item-error branches. A full handler-level mock
across HttpClient/Session is out of scope: the exporter has no such mock
today (its network tests are DISABLED_), and the status invariant is
what the bug was, so it is covered at the validator instead.

Fixes open-telemetry#4295

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

Author: thc1006

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

@@ -28,12 +28,25 @@ namespace detail
  * 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.
+ *
+ * @param status_code the HTTP status code of the response
  * @param body the raw response body
  * @param failure_reason set to a short explanation when the function returns false
- * @return true when the response reports every item as written
+ * @return true when the status is 2xx and the response reports every item as written
  */
-inline bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reason) noexcept
+inline bool IsBulkResponseSuccessful(int status_code,
+                                     const std::string &body,
+                                     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
   {

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -102,6 +102,10 @@ class ResponseHandler : public http_client::EventHandler
                                 << log_message);
       }
 
+      // 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.
+      status_code_ = response.GetStatusCode();
+
       // Set the response_received_ flag to true and notify any threads waiting on this result
       response_received_ = true;
     }
@@ -129,6 +133,15 @@ class ResponseHandler : public http_client::EventHandler
     return body_;
   }
 
+  /**
+   * Returns the HTTP status code of the response
+   */
+  int GetStatusCode()
+  {
+    std::unique_lock<std::mutex> lk(mutex_);
+    return status_code_;
+  }
+
   // Callback method when an http event occurs
   void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override
   {
@@ -201,6 +214,9 @@ class ResponseHandler : public http_client::EventHandler
   // A string to store the response body
   std::string body_ = "";
 
+  // The HTTP status code of the response
+  int status_code_ = 0;
+
   // Whether to print the results from the callback
   bool console_debug_ = false;
 };
@@ -247,17 +263,8 @@ class AsyncResponseHandler : public http_client::EventHandler
       OTEL_INTERNAL_LOG_DEBUG(
           "[ES Log Exporter] Got response from Elasticsearch,  response body: " << body_);
     }
-    const auto status_code = response.GetStatusCode();
-    if (!(status_code >= 200 && status_code <= 299))
-    {
-      OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Export failed with HTTP status "
-                              << status_code << ", response body: " << body_);
-      result_callback_(sdk::common::ExportResult::kFailure);
-      return;
-    }
-
     std::string failure_reason;
-    if (!detail::IsBulkResponseSuccessful(body_, 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_);
@@ -460,7 +467,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(responseBody, 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,36 +164,60 @@ constexpr const char *kOneItemRejected =
 TEST(ElasticsearchBulkResponseTests, ReportsSuccessWhenErrorsIsFalse)
 {
   std::string reason;
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(kPrettySuccess, reason));
-  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(kCompactSuccess, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kPrettySuccess, reason));
+  EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, kCompactSuccess, reason));
 }
 
 TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenAnyItemWasRejected)
 {
   std::string reason;
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(kOneItemRejected, 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("not json at all", reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(200, "not json at all", reason));
   EXPECT_FALSE(reason.empty());
 
-  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful("", 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("[1,2,3]", 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(R"({"took":30,"items":[]})", reason));
-  EXPECT_FALSE(
-      logs_exporter::detail::IsBulkResponseSuccessful(R"({"errors":"false","items":[]})", reason));
+      logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"took":30,"items":[]})", reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
+      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;
+  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));
+}
+
+// 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));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
+      200, R"({"errors":true,"items":[null]})", reason));
+  EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(
+      200, R"({"errors":true,"items":[{"index":42}]})", reason));
 }