From 30e2f322c2d6cf1148bc90c8a9fedae34987a8b9 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 01:19:32 +0800 Subject: [PATCH 1/9] [BUG] Parse the Elasticsearch bulk response instead of matching a substring Both export paths decided whether a bulk write succeeded with body.find("\"failed\" : 0") == std::string::npos which is wrong in both directions. "failed" is per-shard information for a single item, so a batch where one item was rejected and another reported "failed": 0 was recorded as a success and the rejected records were silently lost. The literal also contains pretty-printing spaces, so a compact response never matched and a completely successful export was reported as a failure. Both paths now parse the body and check the top level "errors" field, which is what the bulk API defines as the outcome for the batch, and report the first item error rather than only that something went wrong. A malformed or unparseable body counts as a failure. The asynchronous path also checks the HTTP status, which it previously ignored while the synchronous path did not. nlohmann/json.hpp was already included by this file and already linked in both the CMake and Bazel targets, so this adds no dependency. Fixes #4295 Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../src/es_log_record_exporter.cc | 92 +++++++++++++++++-- 1 file changed, 84 insertions(+), 8 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 8cb7d8140c..531a17df6a 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -41,6 +41,73 @@ namespace exporter { namespace logs { +namespace +{ + +/** + * 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. + * + * @param body the raw response body + * @param failure_reason set to a short explanation when the function returns false + */ +bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reason) +{ + 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; + } + + if (!errors->get()) + { + 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) + { + if (!item.is_object()) + { + continue; + } + for (auto operation = item.begin(); operation != item.end(); ++operation) + { + 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 = "the response reports errors"; + return false; +} + +} // namespace + /** * This class handles the response message from the Elasticsearch request */ @@ -245,11 +312,20 @@ class AsyncResponseHandler : public http_client::EventHandler OTEL_INTERNAL_LOG_DEBUG( "[ES Log Exporter] Got response from Elasticsearch, response body: " << body_); } - if (body_.find("\"failed\" : 0") == std::string::npos) + 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 (!IsBulkResponseSuccessful(body_, failure_reason)) { - OTEL_INTERNAL_LOG_ERROR( - "[ES Log Exporter] Logs were not written to Elasticsearch correctly, response body: " - << body_); + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, " + << failure_reason << ", response body: " << body_); result_callback_(sdk::common::ExportResult::kFailure); } else @@ -448,11 +524,11 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( // Parse the response output to determine if Elasticsearch consumed it correctly std::string responseBody = handler->GetResponseBody(); - if (responseBody.find("\"failed\" : 0") == std::string::npos) + std::string failure_reason; + if (!IsBulkResponseSuccessful(responseBody, failure_reason)) { - OTEL_INTERNAL_LOG_ERROR( - "[ES Log Exporter] Logs were not written to Elasticsearch correctly, response body: " - << responseBody); + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, " + << failure_reason << ", response body: " << responseBody); // TODO: Retry logic return sdk::common::ExportResult::kFailure; } From 6e1713b0a6bebc1ae797fafb37cabe278b635667 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 01:25:05 +0800 Subject: [PATCH 2/9] Add CHANGELOG entry for #4297 Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8baba4503c..d7a50ade5f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,9 @@ Increment the: namespaces [#4303](https://github.com/open-telemetry/opentelemetry-cpp/pull/4303) +* [BUG] Parse the Elasticsearch bulk response instead of matching a substring + [#4297](https://github.com/open-telemetry/opentelemetry-cpp/pull/4297) + * [CODE HEALTH] Move metrics storage test fixtures into anonymous namespace [#4286](https://github.com/open-telemetry/opentelemetry-cpp/pull/4286) From 2856c4b393bfd27a807149a0f9029087e337a15a Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 02:12:55 +0800 Subject: [PATCH 3/9] Keep the bulk response check from throwing into noexcept callers AsyncResponseHandler::OnResponse is noexcept, and the helper it now calls can throw: dump() and the string concatenation that builds the failure reason both allocate. clang-tidy reported bugprone-exception-escape on OnResponse, which was a warning this branch added rather than one that was already there. The helper is now noexcept and absorbs anything thrown while inspecting the body, which is the right answer anyway: a response that cannot be inspected is not a successful export. The default reason is set inside the try so the handler itself does no allocation. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../src/es_log_record_exporter.cc | 79 +++++++++++-------- 1 file changed, 45 insertions(+), 34 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 531a17df6a..a923ff10e9 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -55,55 +55,66 @@ namespace * @param body the raw response body * @param failure_reason set to a short explanation when the function returns false */ -bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reason) +bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reason) noexcept { - const nlohmann::json parsed = nlohmann::json::parse(body, nullptr, false); - if (parsed.is_discarded() || !parsed.is_object()) + // Callers include noexcept response handlers, so nothing may escape from here. Anything that + // stops the body being inspected counts as a failed export. + try { - failure_reason = "the response body is not a JSON object"; - return false; - } + failure_reason = "the response body could not be inspected"; - 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; - } + 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; + } - if (!errors->get()) - { - return true; - } + 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; + } - // 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) + if (!errors->get()) { - if (!item.is_object()) - { - continue; - } - for (auto operation = item.begin(); operation != item.end(); ++operation) + 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) { - 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; + } } } } - } - failure_reason = "the response reports errors"; - return false; + failure_reason = "the response reports errors"; + return false; + } + catch (...) + { + return false; + } } } // namespace From 6d1fab06eff42724c3cb6eb1613e0df57f393c57 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 02:44:01 +0800 Subject: [PATCH 4/9] Guard the try/catch for exception-free builds The Bazel noexcept configuration compiles with -fno-exceptions, where try is a hard error rather than a warning, so the previous commit broke that job. The repository already has an idiom for this, OPENTELEMETRY_HAVE_EXCEPTIONS, used the same way in attribute_utils.h. Nothing is lost when it is off: without exceptions an allocation failure terminates rather than unwinding, so there is nothing for the handler to catch in the first place. The macro arrives through opentelemetry/version.h, which includes macros.h with an IWYU pragma export, so no new include is needed. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- exporters/elasticsearch/src/es_log_record_exporter.cc | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index a923ff10e9..93b9a2019c 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -58,9 +58,12 @@ namespace bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reason) noexcept { // Callers include noexcept response handlers, so nothing may escape from here. Anything that - // stops the body being inspected counts as a failed export. + // stops the body being inspected counts as a failed export. With exceptions disabled there is + // nothing to catch, since an allocation failure terminates instead. +#if OPENTELEMETRY_HAVE_EXCEPTIONS try { +#endif failure_reason = "the response body could not be inspected"; const nlohmann::json parsed = nlohmann::json::parse(body, nullptr, false); @@ -110,11 +113,13 @@ bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reas failure_reason = "the response reports errors"; return false; +#if OPENTELEMETRY_HAVE_EXCEPTIONS } catch (...) { return false; } +#endif } } // namespace From d39a9930791d639c3cf9b37506cd30e5a7e96f87 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 03:08:35 +0800 Subject: [PATCH 5/9] Make the bulk response check testable and cover it Codecov reported 0% patch coverage: none of the 34 changed lines were exercised. The reason was structural rather than an oversight about which tests to write. The helper sat in an anonymous namespace inside the .cc, so no test translation unit could reach it, and the existing Elasticsearch tests are almost all DISABLED_ because they need a live server. It is a pure function from a response body to a verdict, so it belongs somewhere a test can call it. It moves to a detail header, following the detail/ pattern already used by ext/http/client/detail/default_factory.h, and is added to the Bazel hdrs so the test target picks it up. Four tests cover it: pretty-printed and compact successes, a batch with one rejected item where the reason must name the underlying error, an unusable body, and a missing or non-boolean errors field. The first two are the cases the old substring check got wrong in each direction. nlohmann/json.hpp is no longer used by the .cc and is removed from it. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- exporters/elasticsearch/BUILD | 1 + .../elasticsearch/detail/es_bulk_response.h | 102 ++++++++++++++++++ .../src/es_log_record_exporter.cc | 88 +-------------- .../test/es_log_record_exporter_test.cc | 55 ++++++++++ 4 files changed, 161 insertions(+), 85 deletions(-) create mode 100644 exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h diff --git a/exporters/elasticsearch/BUILD b/exporters/elasticsearch/BUILD index 71ff5f39d2..d9882bd103 100644 --- a/exporters/elasticsearch/BUILD +++ b/exporters/elasticsearch/BUILD @@ -13,6 +13,7 @@ cc_library( "src/es_log_recordable.cc", ], hdrs = [ + "include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h", "include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h", "include/opentelemetry/exporters/elasticsearch/es_log_recordable.h", ], diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h new file mode 100644 index 0000000000..85d34df23b --- /dev/null +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h @@ -0,0 +1,102 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include + +#include "opentelemetry/version.h" + +OPENTELEMETRY_BEGIN_NAMESPACE +namespace exporter +{ +namespace logs +{ +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. + * + * 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. + * + * @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 + */ +inline bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reason) noexcept +{ +#if OPENTELEMETRY_HAVE_EXCEPTIONS + try + { +#endif + 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()) + { + 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; + } + + if (!errors->get()) + { + 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) + { + if (!item.is_object()) + { + continue; + } + for (auto operation = item.begin(); operation != item.end(); ++operation) + { + 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 = "the response reports errors"; + return false; +#if OPENTELEMETRY_HAVE_EXCEPTIONS + } + catch (...) + { + return false; + } +#endif +} + +} // namespace detail +} // namespace logs +} // namespace exporter +OPENTELEMETRY_END_NAMESPACE diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 93b9a2019c..d3121b8171 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -8,12 +8,12 @@ #include #include // IWYU pragma: keep #include -#include #include #include #include #include +#include "opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h" #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h" #include "opentelemetry/exporters/elasticsearch/es_log_recordable.h" #include "opentelemetry/ext/http/client/detail/default_factory.h" @@ -41,88 +41,6 @@ namespace exporter { namespace logs { -namespace -{ - -/** - * 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. - * - * @param body the raw response body - * @param failure_reason set to a short explanation when the function returns false - */ -bool IsBulkResponseSuccessful(const std::string &body, std::string &failure_reason) noexcept -{ - // 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. -#if OPENTELEMETRY_HAVE_EXCEPTIONS - try - { -#endif - 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()) - { - 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; - } - - if (!errors->get()) - { - 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) - { - if (!item.is_object()) - { - continue; - } - for (auto operation = item.begin(); operation != item.end(); ++operation) - { - 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 = "the response reports errors"; - return false; -#if OPENTELEMETRY_HAVE_EXCEPTIONS - } - catch (...) - { - return false; - } -#endif -} - -} // namespace /** * This class handles the response message from the Elasticsearch request @@ -338,7 +256,7 @@ class AsyncResponseHandler : public http_client::EventHandler } std::string failure_reason; - if (!IsBulkResponseSuccessful(body_, failure_reason)) + if (!detail::IsBulkResponseSuccessful(body_, failure_reason)) { OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, " << failure_reason << ", response body: " << body_); @@ -541,7 +459,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 (!IsBulkResponseSuccessful(responseBody, failure_reason)) + if (!detail::IsBulkResponseSuccessful(responseBody, failure_reason)) { OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Logs were not written to Elasticsearch correctly, " << failure_reason << ", response body: " << responseBody); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index a65c0b4c1c..b884d13bb6 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -3,6 +3,7 @@ #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h" #include "opentelemetry/common/timestamp.h" +#include "opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h" #include "opentelemetry/exporters/elasticsearch/es_log_recordable.h" #include "opentelemetry/logs/severity.h" #include "opentelemetry/nostd/span.h" @@ -142,3 +143,57 @@ TEST(ElasticsearchLogRecordableTests, BasicTests) EXPECT_EQ(actual, expected); } + +// The bulk API reports the outcome for the batch in the top level "errors" field. These cover +// the two cases the previous substring test got wrong: a batch that had a rejected item was +// reported as a success, and a successful batch from a server that does not pretty-print was +// reported as a failure. +namespace +{ +constexpr const char *kPrettySuccess = + R"({"took":30,"errors":false,"items":[{"index":{"_shards":{"total":2,"successful":1,)" + R"("failed" : 0}}}]})"; +constexpr const char *kCompactSuccess = + R"({"took":30,"errors":false,"items":[{"index":{"_shards":{"total":2,"successful":1,)" + R"("failed":0}}}]})"; +constexpr const char *kOneItemRejected = + R"({"took":30,"errors":true,"items":[{"index":{"_shards":{"failed" : 0}}},)" + R"({"index":{"error":{"type":"mapper_parsing_exception","reason":"bad field"}}}]})"; +} // namespace + +TEST(ElasticsearchBulkResponseTests, ReportsSuccessWhenErrorsIsFalse) +{ + std::string reason; + EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(kPrettySuccess, reason)); + EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(kCompactSuccess, reason)); +} + +TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenAnyItemWasRejected) +{ + std::string reason; + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(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(reason.empty()); + + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful("", reason)); + EXPECT_FALSE(reason.empty()); + + // An array rather than the expected object. + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful("[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)); +} From d7593536136305a5fb69c2e7a90ce8ea5334ab9e Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 04:10:17 +0800 Subject: [PATCH 6/9] Keep the direct nlohmann include in the exporter IWYU flagged es_log_record_exporter.cc for a missing nlohmann/json.hpp: it calls GetJSON().dump() at the point it builds the request body, which is a direct use of the type. When the bulk response helper moved to its own header I removed the include with it, but this use remained, and IWYU treats reaching it through es_log_recordable.h as an indirect dependency. The include goes back. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- exporters/elasticsearch/src/es_log_record_exporter.cc | 1 + 1 file changed, 1 insertion(+) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index d3121b8171..ee20c55fdc 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -8,6 +8,7 @@ #include #include // IWYU pragma: keep #include +#include #include #include #include From 04357c7b6981d1638d7f5737de655086f5a13981 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 13:40:27 +0800 Subject: [PATCH 7/9] Fold the HTTP status into the bulk export result on both paths Review of #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 #4295 Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../elasticsearch/detail/es_bulk_response.h | 17 +++++++- .../src/es_log_record_exporter.cc | 29 ++++++++----- .../test/es_log_record_exporter_test.cc | 42 +++++++++++++++---- 3 files changed, 66 insertions(+), 22 deletions(-) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h index 85d34df23b..cf474bc1bf 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h +++ b/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 { diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index ee20c55fdc..26f4c0bec3 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/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 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); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index b884d13bb6..c39a671b51 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -164,28 +164,28 @@ 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()); } @@ -193,7 +193,31 @@ TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNot { 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)); } From 47be24f76fbf950c3b518206dfc9c018b77a6535 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 16:00:17 +0800 Subject: [PATCH 8/9] 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> --- .../elasticsearch/detail/es_bulk_response.h | 62 +++++++++++++------ .../src/es_log_record_exporter.cc | 34 +++++----- .../test/es_log_record_exporter_test.cc | 61 +++++++++++++----- 3 files changed, 102 insertions(+), 55 deletions(-) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h index cf474bc1bf..3d5ce77885 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h @@ -3,6 +3,7 @@ #pragma once +#include #include #include @@ -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()) { 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; } } } diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 26f4c0bec3..3776a33630 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -75,8 +75,6 @@ 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 lk(mutex_); @@ -84,26 +82,15 @@ class ResponseHandler : public http_client::EventHandler // 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 session, std::function &&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); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index c39a671b51..1d44a03bcd 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -164,28 +164,28 @@ 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()); } @@ -193,9 +193,9 @@ TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNot { 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 @@ -203,11 +203,12 @@ TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNot 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. @@ -215,9 +216,39 @@ TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsTrueWithoutItemErro { 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)); } From 229ee1902adeab69ed7430657fa081dd26bdd524 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 17:25:42 +0800 Subject: [PATCH 9/9] 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> --- .../elasticsearch/detail/es_bulk_response.h | 81 +++++++------------ .../src/es_log_record_exporter.cc | 13 +-- .../test/es_log_record_exporter_test.cc | 69 ++++++---------- 3 files changed, 57 insertions(+), 106 deletions(-) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h index 3d5ce77885..9fa425f89a 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h @@ -3,7 +3,6 @@ #pragma once -#include #include #include @@ -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()) { 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; + } } } } diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 3776a33630..8d16dd7f5b 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -221,11 +221,9 @@ class AsyncResponseHandler : public http_client::EventHandler AsyncResponseHandler( std::shared_ptr session, std::function &&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); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 1d44a03bcd..a8124e156c 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -164,28 +164,36 @@ 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()); } @@ -193,9 +201,9 @@ TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNot { 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 @@ -203,12 +211,11 @@ TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNot 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. @@ -216,39 +223,9 @@ TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsTrueWithoutItemErro { 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)); }