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) 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..9fa425f89a --- /dev/null +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h @@ -0,0 +1,118 @@ +// 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. 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. + * + * 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 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::string &failure_reason) noexcept +{ + failure_reason.clear(); +#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; + } + + 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; + } + + // 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) + { + 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 8cb7d8140c..8d16dd7f5b 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -14,6 +14,7 @@ #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,6 +42,7 @@ namespace exporter { namespace logs { + /** * This class handles the response message from the Elasticsearch request */ @@ -73,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_); @@ -82,24 +82,17 @@ 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. 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 response_received_ = true; } @@ -127,6 +120,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 { @@ -199,6 +201,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; }; @@ -245,11 +250,11 @@ 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) + std::string failure_reason; + if (!detail::IsBulkResponseSuccessful(response.GetStatusCode(), 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 @@ -368,7 +373,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"); @@ -448,11 +453,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 (!detail::IsBulkResponseSuccessful(handler->GetStatusCode(), 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; } diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index a65c0b4c1c..a8124e156c 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,89 @@ 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(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, 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(reason.empty()); + + 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]", reason)); + EXPECT_FALSE(reason.empty()); +} + +TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNotBoolean) +{ + std::string reason; + EXPECT_FALSE( + 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)); +}