From 3fc79c5fc11bae2da4b45bb1a66c2130bfc2a0e5 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Tue, 8 Sep 2026 05:48:04 +0000 Subject: [PATCH 1/2] [BUG] Decide Elasticsearch bulk export success from status and errors Both export paths decided whether a bulk write had succeeded by searching the response body for the literal `"failed" : 0`. That field is per-shard information for a single item rather than the outcome of the batch, which is the top-level `errors` flag, and the literal baked in pretty-printing whitespace besides. It was wrong in both directions: a batch holding a rejected item read as success, and a compact successful response read as failure. One helper, `IsBulkResponseSuccessful(status_code, body, expected_items, reason)`, now answers for both paths. A non-2xx status fails first. Then the body has to be a JSON object carrying a boolean `errors` flag and an `items` array with one operation result per operation the request submitted, and `errors` has to be false. A malformed body is a failure rather than a silent success, and the reason the helper reports is the one written to the log. Squashed from fourteen commits so the rebase onto main resolves once rather than at every step. No content change: the tree is identical to the one the fourteen produced. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 5 + exporters/elasticsearch/BUILD | 20 + exporters/elasticsearch/CMakeLists.txt | 12 + .../elasticsearch/detail/es_bulk_response.h | 208 ++++++ .../src/es_log_record_exporter.cc | 101 ++- .../test/es_log_record_exporter_test.cc | 703 ++++++++++++++++++ 6 files changed, 1009 insertions(+), 40 deletions(-) create mode 100644 exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h diff --git a/CHANGELOG.md b/CHANGELOG.md index 44bfbcf494..42623b9651 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -210,6 +210,11 @@ Increment the: * [BUG] Wait on a completion state in the Elasticsearch exporter [#4298](https://github.com/open-telemetry/opentelemetry-cpp/pull/4298) +* [BUG] Decide Elasticsearch bulk export success from the HTTP status, the + errors flag, and one acknowledgement per record that names its target index + and says the operation applied + [#4297](https://github.com/open-telemetry/opentelemetry-cpp/pull/4297) + * [BUG] Make SocketAddr string parsing safe and reject malformed addresses [#4292](https://github.com/open-telemetry/opentelemetry-cpp/pull/4292) diff --git a/exporters/elasticsearch/BUILD b/exporters/elasticsearch/BUILD index 71ff5f39d2..967aad14d8 100644 --- a/exporters/elasticsearch/BUILD +++ b/exporters/elasticsearch/BUILD @@ -6,6 +6,24 @@ load("@rules_cc//cc:cc_test.bzl", "cc_test") package(default_visibility = ["//visibility:public"]) +# The bulk response parser is an implementation detail that the CMake package +# also leaves out. hdrs is a target's public interface, so it is kept in its own +# target that only this package can depend on, and reached through +# implementation_deps so the exporter does not re-export it. +cc_library( + name = "es_bulk_response", + hdrs = [ + "include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h", + ], + strip_include_prefix = "include", + tags = ["es"], + visibility = ["//exporters/elasticsearch:__pkg__"], + deps = [ + "//api", + "@github_nlohmann_json//:json", + ], +) + cc_library( name = "es_log_record_exporter", srcs = [ @@ -16,6 +34,7 @@ cc_library( "include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h", "include/opentelemetry/exporters/elasticsearch/es_log_recordable.h", ], + implementation_deps = [":es_bulk_response"], linkopts = select({ "//bazel:windows": [ "-DEFAULTLIB:advapi32.lib", @@ -43,6 +62,7 @@ cc_test( "test", ], deps = [ + ":es_bulk_response", ":es_log_record_exporter", "@com_google_googletest//:gtest_main", "@curl", diff --git a/exporters/elasticsearch/CMakeLists.txt b/exporters/elasticsearch/CMakeLists.txt index 18b2bcd898..9f900cf36c 100644 --- a/exporters/elasticsearch/CMakeLists.txt +++ b/exporters/elasticsearch/CMakeLists.txt @@ -34,6 +34,12 @@ otel_add_component( "*.h" PATTERN "es_log_recordable.h" + EXCLUDE + PATTERN + "es_bulk_response.h" + EXCLUDE + PATTERN + "detail" EXCLUDE) if(OTELCPP_INSTALL) @@ -55,4 +61,10 @@ if(OTELCPP_BUILD_TESTING) TARGET es_log_record_exporter_test TEST_PREFIX exporter. TEST_LIST es_log_record_exporter_test) + + # The synchronous wiring cases wait on a completion, and a regression that + # stops delivering one hangs the job rather than reporting. A bound turns that + # back into a red test. Not the asynchronous ones: their fake calls back from + # inside SendRequest() and Export() returns without waiting. + set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 120) endif() # OTELCPP_BUILD_TESTING 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..d51b598794 --- /dev/null +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h @@ -0,0 +1,208 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include + +#include "opentelemetry/version.h" + +OPENTELEMETRY_BEGIN_NAMESPACE +namespace exporter +{ +namespace logs +{ +namespace detail +{ + +/** + * What the bulk request itself has to answer with. Elasticsearch answers a bulk request with 200, + * and the rest of the 2xx band does not say the batch was written. 202 is the one that matters: + * it says the request was accepted and the processing is not finished, and may never be. Reading + * that as success is not a conservative reading of a vague answer, it is the opposite of what the + * answer says, and a success here lets the caller drop the records it just handed over. + */ +inline bool IsSuccessfulBulkStatus(int status_code) noexcept +{ + return 200 == status_code; +} + +/** + * Whether an acknowledged operation status is one that applied the operation. Elasticsearch + * answers 201 when it created the document and 200 when it replaced an existing one. + * + * Comparing the json value rather than extracting one keeps this noexcept without relying on the + * type check to make an extraction safe. The check still has to be here: without it a float 200.5 + * would answer for 200. + */ +inline bool IsAcknowledgedStatus(const nlohmann::json &status) noexcept +{ + return status.is_number_integer() && (status == 200 || status == 201); +} + +/** + * Decide whether an Elasticsearch bulk response reports the whole batch as written. + * + * Callers include noexcept response handlers, so nothing may escape. Anything that stops the body + * being inspected counts as a failed export. + * + * This reads the fields the bulk response documents to decide an outcome. It is not a check + * against a responder that is trying to be believed: a repeated key, for one, is folded before + * this sees the document, so a body carrying both "errors": true and "errors": false arrives as + * whichever one the parser kept, and nothing here can tell that the other was ever sent. + * + * @param status_code the response status, which the caller passes rather than this reading the + * body alone: "errors" describes item outcomes and cannot override a transport error + * @param body the raw response body + * @param expected_items the number of index operations the request submitted + * @param failure_reason a best-effort explanation when this returns false + * @return true only when the status is 200, "errors" is false, and "items" holds exactly + * expected_items index results that each name a target index, acknowledge a status the + * operation applied under, and carry no error + */ +inline bool IsBulkResponseSuccessful(int status_code, + const std::string &body, + std::size_t expected_items, + std::string &failure_reason) noexcept +{ + failure_reason.clear(); +#if OPENTELEMETRY_HAVE_EXCEPTIONS + try + { +#endif + // Inside the try: building a reason allocates. + // Not the band an operation status is held to. This one answers for the request, and an + // operation that applied answers 201 for a document it created, which the request itself + // never says. + if (!IsSuccessfulBulkStatus(status_code)) + { + 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; + } + + // The request is an unfiltered /_bulk, so "items" is always there and holds one entry per + // operation. The next three checks decide whether this body answers the request that was sent. + 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; + } + + // Each entry is keyed by the action it answers, and every record goes out as an index + // operation. Read before "errors", which a body that is not this answer does not get to decide. + // An operation that was not applied says so twice, through its status and through an "error" + // member. Both are read here so that neither reading depends on the "errors" flag, which a + // response that contradicts itself also controls. + const nlohmann::json *rejected = nullptr; + const nlohmann::json *item_error = nullptr; + for (const auto &item : *items) + { + if (!item.is_object() || item.size() != 1) + { + failure_reason = "the response has an \"items\" entry that is not one operation result"; + return false; + } + + const auto operation = item.find("index"); + if (operation == item.end() || !operation->is_object()) + { + failure_reason = "the response does not acknowledge the submitted index operation"; + return false; + } + + // Elasticsearch names the index it wrote to in every index result, whether the operation + // applied or not. Presence and type only: the name it reports is the index the write + // resolved to, which an alias or a date math index makes different from the one that was + // submitted, so it is not something to compare against. + const auto target = operation->find("_index"); + if (target == operation->end() || !target->is_string()) + { + failure_reason = "the response acknowledges an index operation with no target index"; + return false; + } + + const auto status = operation->find("status"); + if (status == operation->end() || !status->is_number_integer()) + { + failure_reason = "the response acknowledges an index operation with no status"; + return false; + } + + if (rejected == nullptr && !IsAcknowledgedStatus(*status)) + { + rejected = &(*status); + } + + // A null holds no cause, and a serialiser that writes absent optionals as null is saying + // the operation applied, which is what the flag says too. Only a cause contradicts it. + const auto error = operation->find("error"); + if (item_error == nullptr && error != operation->end() && !error->is_null()) + { + item_error = &(*error); + } + } + + if (!errors->get()) + { + // A false flag claims every operation was applied; hold the items to that claim. + if (rejected != nullptr) + { + failure_reason = + "the response reports no errors but acknowledges operation status " + rejected->dump(); + return false; + } + if (item_error != nullptr) + { + failure_reason = + "the response reports no errors but an item carries error " + item_error->dump(); + return false; + } + return true; + } + + // Name the first item error rather than only saying that something failed. + if (item_error != nullptr) + { + failure_reason = "at least one item failed, first error: " + item_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 cd3b1bdbfd..55ab2e8eba 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" @@ -75,37 +76,42 @@ class ResponseHandler : public http_client::EventHandler */ void OnResponse(http_client::Response &response) noexcept override { - std::string log_message; + std::string described; + bool decided = false; // Lock the private members so they can't be read while being modified { std::unique_lock lk(mutex_); - // Store the body of the request - body_ = std::string(response.GetBody().begin(), response.GetBody().end()); + std::string body(response.GetBody().begin(), response.GetBody().end()); - if (!(response.GetStatusCode() >= 200 && response.GetStatusCode() <= 299)) + // Kept with the outcome it decides, and only by the call that decides it. Export() folds the + // status into the ExportResult, since storing the body alone let a non-2xx response be + // reported as a success, and a status from one response beside a body from another is not + // something the server sent. 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. + if (completion_ == CompletionState::Pending) { - log_message = BuildResponseLogMessage(response, body_); - - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Export failed, " << log_message); - } - - if (console_debug_) - { - if (log_message.empty()) + if (console_debug_) { - log_message = BuildResponseLogMessage(response, body_); + described = BuildResponseLogMessage(response, body); } - - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Got response from Elasticsearch, " - << log_message); + status_code_ = response.GetStatusCode(); + body_ = std::move(body); + recordCompletionLocked(CompletionState::Success); + decided = true; } - - // Record the outcome and notify any threads waiting on this result - recordCompletionLocked(CompletionState::Success); } cv_.notify_all(); + + // Outside the lock and only for the response that decided the outcome. The log handler is + // replaceable application code, so one that reaches back into this exporter would otherwise do + // it while this thread holds mutex_, and a line for a response whose status and body were + // discarded describes an answer the caller was never given. + if (decided && console_debug_) + { + OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Got response from Elasticsearch, " << described); + } } /** @@ -122,14 +128,21 @@ class ResponseHandler : public http_client::EventHandler return completion_ == CompletionState::Success; } + /// The status and body of the response the recorded outcome was decided from. + struct Response + { + int status_code = 0; + std::string body; + }; + /** - * Returns the body of the response + * Returns the response the outcome was decided from. Both halves together, since reading them + * one lock at a time can pair a status with a body that arrived in a different response. */ - std::string GetResponseBody() + Response GetResponse() { - // Lock so that body_ can't be written to while returning it std::unique_lock lk(mutex_); - return body_; + return Response{status_code_, body_}; } // Callback method when an http event occurs @@ -236,6 +249,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; }; @@ -253,9 +269,11 @@ class AsyncResponseHandler : public http_client::EventHandler AsyncResponseHandler( std::shared_ptr session, std::function &&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} {} @@ -275,18 +293,20 @@ class AsyncResponseHandler : public http_client::EventHandler void OnResponse(http_client::Response &response) noexcept override { - // Store the body of the response - body_ = std::string(response.GetBody().begin(), response.GetBody().end()); + // Local, since nothing outside this call reads it and a member would be written twice over by + // a client that delivers two responses for one request. + const std::string body(response.GetBody().begin(), response.GetBody().end()); if (console_debug_) { OTEL_INTERNAL_LOG_DEBUG( - "[ES Log Exporter] Got response from Elasticsearch, response body: " << body_); + "[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, submitted_operations_, + 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 @@ -344,8 +364,8 @@ class AsyncResponseHandler : public http_client::EventHandler // Callback to call to on receiving events std::function result_callback_; - // A string to store the response body - std::string body_ = ""; + // How many operations this request submitted, which is how many the response has to answer + std::size_t submitted_operations_ = 0; // Whether to print the results from the callback bool console_debug_ = false; @@ -406,7 +426,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"); @@ -423,7 +443,7 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( for (auto &record : records) { // Append {"index":{}} before JSON body, which tells Elasticsearch to write to index specified - // in URI + // in URI. detail/es_bulk_response.h requires each acknowledgement to name this same action. body += "{\"index\" : {}}\n"; // Add the context of the Recordable @@ -464,7 +484,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 @@ -491,12 +511,13 @@ 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) + const auto response = handler->GetResponse(); + std::string failure_reason; + if (!detail::IsBulkResponseSuccessful(response.status_code, response.body, records.size(), + 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: " << response.body); // 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 52a5d9b6ef..843340ecd2 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -3,14 +3,17 @@ #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/ext/http/client/http_client.h" #include "opentelemetry/logs/severity.h" #include "opentelemetry/nostd/function_ref.h" +#include "opentelemetry/nostd/shared_ptr.h" #include "opentelemetry/nostd/span.h" #include "opentelemetry/nostd/string_view.h" #include "opentelemetry/nostd/utility.h" #include "opentelemetry/sdk/common/exporter_utils.h" +#include "opentelemetry/sdk/common/global_log_handler.h" #include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h" #include "opentelemetry/sdk/logs/exporter.h" #include "opentelemetry/sdk/logs/recordable.h" @@ -21,8 +24,10 @@ #include #include #include +#include #include #include +#include #include "nlohmann/json.hpp" namespace @@ -263,3 +268,701 @@ TEST(ElasticsearchLogRecordableTests, BasicTests) EXPECT_EQ(actual, expected); } + +// The batch outcome is the top level "errors" field. Two success bodies that differ only in +// whitespace, so a check that depends on formatting cannot answer the same for both. +namespace +{ +constexpr const char *kPrettySuccess = + R"({"took":30,"errors":false,"items":[{"index":{"_index":"logs","_id":"1",)" + R"("_shards":{"total":2,"successful":1,"failed" : 0},"status":201}}]})"; +// Two operations acknowledged, so a batch of two has something to match. 201 for a new document +// and 200 for an overwrite are both what Elasticsearch answers for an index operation. +constexpr const char *kTwoItemCompactSuccess = R"({"took":30,"errors":false,"items":[)" + R"({"index":{"_index":"logs","status":201}},)" + R"({"index":{"_index":"logs","status":200}}]})"; + +constexpr const char *kCompactSuccess = + R"({"took":30,"errors":false,"items":[{"index":{"_index":"logs","_id":"1",)" + R"("_shards":{"total":2,"successful":1,"failed":0},"status":201}}]})"; +constexpr const char *kOneItemRejected = + R"({"took":30,"errors":true,"items":[{"index":{"_index":"logs","_id":"1",)" + R"("_shards":{"failed" : 0},"status":201}},{"index":{"_index":"logs","_id":"2",)" + R"("status":400,"error":{"type":"mapper_parsing_exception","reason":"bad field"}}}]})"; +} // namespace + +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)); +} + +// 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, 1, reason)); + EXPECT_TRUE(reason.empty()); +} + +TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenAnyItemWasRejected) +{ + std::string 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", 1, reason)); + EXPECT_FALSE(reason.empty()); + + 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]", 1, reason)); + EXPECT_FALSE(reason.empty()); +} + +TEST(ElasticsearchBulkResponseTests, ReportsFailureWhenErrorsFieldIsMissingOrNotBoolean) +{ + std::string reason; + EXPECT_FALSE( + logs_exporter::detail::IsBulkResponseSuccessful(200, R"({"took":30,"items":[]})", 1, reason)); + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, R"({"errors":"false","items":[]})", 0, reason)); +} + +// A non-2xx status is a failure even when the body reports errors:false. This is the invariant +// the substring check and the body-only check both missed, on both the sync and async paths. +TEST(ElasticsearchBulkResponseTests, ReportsFailureOnNon2xxStatus) +{ + std::string reason; + 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)); +} + +// 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":[]})", + 0, reason)); + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, R"({"errors":true,"items":[null]})", 1, reason)); + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, R"({"errors":true,"items":[{"index":42}]})", 1, reason)); +} + +// 200 is what Elasticsearch answers a bulk request with, and it is the only status that says the +// batch was written. 202 is the one worth naming: it says the request was accepted and the +// processing is not finished, so reading it as success would let the caller drop records the +// server has not promised to keep. This pins the boundary on both sides. +TEST(ElasticsearchBulkResponseTests, AcceptsOnlyTheStatusThatSaysTheBatchWasWritten) +{ + std::string reason; + const char *ok_body = R"({"errors":false,"items":[]})"; + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(199, ok_body, 0, reason)); + EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful(200, ok_body, 0, reason)); + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(201, ok_body, 0, reason)); + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(202, ok_body, 0, reason)) + << "202 says the processing is not finished, which is not a write acknowledgement"; + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(299, ok_body, 0, reason)); + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful(300, ok_body, 0, reason)); +} + +// 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; + 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":{"_index":"logs","_shards":{"failed":0},"status":201}}]})"; + constexpr const char *kTwoItemBody = + R"({"errors":false,"items":[{"index":{"_index":"logs","status":201}},{"index":{"_index":"logs","status":201}}]})"; + + 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 right count of the wrong entries is still the wrong answer, and the verdict cannot depend on +// "errors", which the same responder controls. +TEST(ElasticsearchBulkResponseTests, RejectsItemsEntriesThatDoNotAcknowledgeAnIndexOperation) +{ + std::string reason; + const auto rejected = [&reason](const char *body, std::size_t expected) { + return !logs_exporter::detail::IsBulkResponseSuccessful(200, body, expected, reason); + }; + + EXPECT_TRUE(rejected(R"({"errors":false,"items":[null,null]})", 2)); + EXPECT_FALSE(reason.empty()); + EXPECT_TRUE(rejected(R"({"errors":false,"items":[1,2]})", 2)); + EXPECT_TRUE(rejected(R"({"errors":false,"items":[[]]})", 1)); + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{}]})", 1)) << "no operation at all"; + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{"unknown":{"status":201}}]})", 1)) + << "an operation the exporter never submitted"; + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{"index":42}]})", 1)) + << "the index key holds no result object"; + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{"index":{},"delete":{}}]})", 1)) + << "one entry cannot answer two operations"; + EXPECT_TRUE(rejected( + R"({"errors":false,"items":[{"index":{"_index":"logs","status":201},"delete":{}}]})", 1)) + << "a valid acknowledgement does not license a second member"; + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{"index":{"_index":"logs"}}]})", 1)) + << "an index result carrying no status"; + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{"index":{"status":201}}]})", 1)) + << "an index result naming no target index"; + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{"index":{"_index":7,"status":201}}]})", 1)) + << "a target index that is not a name"; + EXPECT_TRUE(rejected(R"({"errors":false,"items":[{"index":{"_index":null,"status":201}}]})", 1)) + << "a target index that is not there under another spelling"; + EXPECT_TRUE(rejected(R"({"errors":true,"items":[null,null]})", 2)) + << "the same shape has to fail whichever way errors reads"; + + // The shape a real acknowledgement has: one index result, naming the index it wrote to and + // carrying the status it wrote with. + EXPECT_FALSE( + rejected(R"({"errors":false,"items":[{"index":{"_index":"logs","status":201}}]})", 1)); + + // Three shapes, three reasons: they are different things to go and look at. + const auto reason_for = [&reason](const char *body) { + logs_exporter::detail::IsBulkResponseSuccessful(200, body, 1, reason); + return reason; + }; + const std::string padded = reason_for(R"({"errors":false,"items":[{"index":{},"delete":{}}]})"); + const std::string other = reason_for(R"({"errors":false,"items":[{"unknown":{"status":1}}]})"); + const std::string no_stat = reason_for(R"({"errors":false,"items":[{"index":{"_index":"l"}}]})"); + const std::string no_idx = reason_for(R"({"errors":false,"items":[{"index":{"status":201}}]})"); + EXPECT_NE(padded, other) << padded; + EXPECT_NE(other, no_stat) << other; + EXPECT_NE(padded, no_stat) << no_stat; + EXPECT_NE(no_stat, no_idx) << no_idx; +} + +// The status says whether the operation applied, so the band has to hold for the number the server +// sent rather than for what it becomes on the way into an int. +TEST(ElasticsearchBulkResponseTests, RejectsAcknowledgementsOutsideTheSuccessBand) +{ + std::string reason; + const auto applied = [&reason](const char *status) { + const std::string body = + std::string(R"({"errors":false,"items":[{"index":{"_index":"logs","status":)") + status + + R"(}}]})"; + return logs_exporter::detail::IsBulkResponseSuccessful(200, body, 1, reason); + }; + + EXPECT_FALSE(applied("199")); + EXPECT_TRUE(applied("200")) << "an index operation that replaced a document"; + EXPECT_TRUE(applied("201")) << "an index operation that created one"; + EXPECT_FALSE(applied("202")) << "accepted is not applied"; + EXPECT_FALSE(applied("299")); + EXPECT_FALSE(applied("300")); + + EXPECT_FALSE(applied("0")) << "zero is a status, not the absence of one"; + EXPECT_FALSE(reason.empty()); + EXPECT_FALSE(applied("-0")); + EXPECT_FALSE(applied("-1")); + EXPECT_FALSE(applied("199")); + EXPECT_FALSE(applied("300")); + EXPECT_FALSE(applied("4294967496")) << "2^32 + 200, which lands on 200 in a 32 bit int"; + EXPECT_FALSE(applied("18446744073709551615")) << "no signed type holds it"; + EXPECT_FALSE(applied("200.5")) << "not an integer"; + EXPECT_FALSE(applied(R"("200")")) << "a string is not a status"; + EXPECT_FALSE(applied("null")); + + // The value reaches the log as it arrived, so a reader sees what the server said. + applied("4294967496"); + EXPECT_NE(reason.find("4294967496"), std::string::npos) << reason; +} + +// A false flag claims every operation was applied. No conforming server sends it alongside a +// rejection, so holding the items to the claim rejects nothing the flag alone would have taken. +TEST(ElasticsearchBulkResponseTests, RejectsAnAcknowledgedFailureUnderErrorsFalse) +{ + std::string reason; + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, R"({"errors":false,"items":[{"index":{"_index":"logs","status":400}}]})", 1, reason)); + EXPECT_NE(reason.find("400"), std::string::npos) << "the reason names the status: " << reason; + + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, + R"({"errors":false,"items":[{"index":{"_index":"logs","status":201}},{"index":{"_index":"logs","status":429}}]})", + 2, reason)) + << "one rejected operation among several is still a rejection"; + + // Both of the statuses an index operation answers with when it applied, in one response. + EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, + R"({"errors":false,"items":[{"index":{"_index":"logs","status":200}},{"index":{"_index":"logs","status":201}}]})", + 2, reason)); + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, R"({"errors":false,"items":[{"index":{"_index":"logs","status":300}}]})", 1, reason)); +} + +// An operation that did not apply says so through its status and through an "error" member. The +// errors:true path already reads the second one, so reading it here as well is what stops the same +// body from being accepted or rejected according to a flag the responder also controls. +TEST(ElasticsearchBulkResponseTests, RejectsAnItemErrorUnderErrorsFalse) +{ + std::string reason; + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, + R"({"errors":false,"items":[{"index":{"_index":"logs","status":201,)" + R"("error":{"type":"mapper_parsing_exception","reason":"rejected"}}}]})", + 1, reason)); + EXPECT_NE(reason.find("mapper_parsing_exception"), std::string::npos) + << "the reason names what the server said: " << reason; + + // A conforming server sends the flag and the member together, and that reads the same way. + EXPECT_FALSE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, + R"({"errors":true,"items":[{"index":{"_index":"logs","status":400,)" + R"("error":{"type":"mapper_parsing_exception"}}}]})", + 1, reason)); + EXPECT_NE(reason.find("mapper_parsing_exception"), std::string::npos) << reason; + + // Only a cause decides. An acknowledgement without the member stays a success, and so does one + // whose member is null, which is what a serialiser writing absent optionals produces. + EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, R"({"errors":false,"items":[{"index":{"_index":"logs","status":201}}]})", 1, reason)); + EXPECT_TRUE(logs_exporter::detail::IsBulkResponseSuccessful( + 200, R"({"errors":false,"items":[{"index":{"_index":"logs","status":201,"error":null}}]})", 1, + reason)); +} +// --------------------------------------------------------------------------- +// The response travelling from the HTTP callback to the verdict. +// +// The cases above call IsBulkResponseSuccessful() directly, so they cannot show +// that the status and the body actually reach it. A fake HTTP client drives the +// callbacks from inside SendRequest(), which runs before Export() reaches the +// wait, and the assertions are on the ExportResult. +// --------------------------------------------------------------------------- +namespace +{ +namespace http_client = opentelemetry::ext::http::client; + +class FakeResponse : public http_client::Response +{ +public: + FakeResponse(http_client::StatusCode status, const std::string &body) + : status_(status), body_(body.begin(), body.end()) + {} + const http_client::Body &GetBody() const noexcept override { return body_; } + bool ForEachHeader( + nostd::function_ref) const noexcept override + { + return true; + } + bool ForEachHeader( + const nostd::string_view &, + nostd::function_ref) const noexcept override + { + return true; + } + http_client::StatusCode GetStatusCode() const noexcept override { return status_; } + +private: + http_client::StatusCode status_; + http_client::Body body_; +}; + +class FakeRequest : public http_client::Request +{ +public: + void SetMethod(http_client::Method) noexcept override {} + void SetUri(nostd::string_view) noexcept override {} + void SetSslOptions(const http_client::HttpSslOptions &) noexcept override {} + void SetBody(http_client::Body &) noexcept override {} + void AddHeader(nostd::string_view, nostd::string_view) noexcept override {} + void ReplaceHeader(nostd::string_view, nostd::string_view) noexcept override {} + void SetTimeoutMs(std::chrono::milliseconds) noexcept override {} + void SetCompression(const http_client::Compression &) noexcept override {} + void EnableLogging(bool) noexcept override {} + void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {} +}; + +using EventScript = std::function; + +class FakeSession : public http_client::Session +{ +public: + explicit FakeSession(EventScript script) : script_(std::move(script)) {} + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + void SendRequest(std::shared_ptr handler) noexcept override + { + script_(*handler); + } + bool IsSessionActive() noexcept override { return false; } + bool CancelSession() noexcept override { return true; } + bool FinishSession() noexcept override { return true; } + +private: + EventScript script_; +}; + +class FakeHttpClient : public http_client::HttpClient +{ +public: + explicit FakeHttpClient(EventScript script) : script_(std::move(script)) {} + std::shared_ptr CreateSession(nostd::string_view) noexcept override + { + return std::make_shared(script_); + } + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + EventScript script_; +}; + +// The response has to answer one operation per record, so a case using a body with N items has to +// export N records or it would be rejected on the count before reaching what it means to test. +opentelemetry::sdk::common::ExportResult ExportWith(EventScript script, + std::size_t records = 1, + bool console_debug = false) +{ + auto client = std::make_shared(std::move(script)); + logs_exporter::ElasticsearchExporterOptions options; + options.console_debug_ = console_debug; + logs_exporter::ElasticsearchLogRecordExporter exporter(options, client); + + std::vector> batch; + batch.reserve(records); + for (std::size_t i = 0; i < records; ++i) + { + batch.push_back(exporter.MakeRecordable()); + } + return exporter.Export( + nostd::span>(batch.data(), batch.size())); +} +} // namespace + +// The synchronous wait exists only when the exporter is built without async export, so these cases +// skip rather than compile out: gtest_add_tests reads the source, and a case that disappeared from +// the binary would still be registered with CTest. The skip goes in SetUp rather than at the top of +// each body, because GTEST_SKIP returns and leaves the rest of the body unreachable, which MSVC +// reports as C4702 and the maintainer mode jobs turn into an error. +namespace +{ +class ElasticsearchLogsExporterWiringTests : public ::testing::Test +{ +protected: + void SetUp() override + { +#ifdef ENABLE_ASYNC_EXPORT + GTEST_SKIP() << "Export() returns without waiting when async export is enabled"; +#endif + } +}; +} // namespace + +// A body the parser accepts, through the whole path rather than through the helper alone. +TEST_F(ElasticsearchLogsExporterWiringTests, AcceptedBulkResponseIsASuccessfulExport) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + FakeResponse response(200, kPrettySuccess); + handler.OnResponse(response); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); +} + +// The first response decides the outcome, so it has to be the one the verdict is read from. A +// client that answers twice would otherwise leave the decision with the first and the evidence +// for it with the second. +TEST_F(ElasticsearchLogsExporterWiringTests, ASecondResponseDoesNotReplaceTheOneThatDecided) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + FakeResponse rejected(500, kPrettySuccess); + handler.OnResponse(rejected); + FakeResponse accepted(200, kPrettySuccess); + handler.OnResponse(accepted); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure) + << "the 500 that decided the outcome was read back as the 200 that followed it"; +} + +// The case #4295 reports: a 200 whose rejected item still leaves a shard counter reading +// "failed" : 0, so the body looks successful to anything that reads it as text. +TEST_F(ElasticsearchLogsExporterWiringTests, RejectedItemIsAFailedExport) +{ + // Two records, because kOneItemRejected answers two operations. Exporting one would be rejected + // on the acknowledgement count and the case would stop testing what it is named for. + const auto result = ExportWith( + [](http_client::EventHandler &handler) { + FakeResponse response(200, kOneItemRejected); + handler.OnResponse(response); + }, + 2); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); +} + +// The status has to reach the parser, not just the body. A handler that stored a fixed status, +// or an Export() that never asked for it, would still pass every case above. +// The count has to travel. Every other successful case here sends one record, so an expected +// count fixed at one would satisfy them all, and the only two item case is a rejection that fails +// whatever the count is. +TEST_F(ElasticsearchLogsExporterWiringTests, TwoAcceptedRecordsUseTheActualBatchSize) +{ + const auto result = ExportWith( + [](http_client::EventHandler &handler) { + FakeResponse response(200, kTwoItemCompactSuccess); + handler.OnResponse(response); + }, + 2); + + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); +} + +TEST_F(ElasticsearchLogsExporterWiringTests, ServerErrorIsAFailedExportEvenWithAnAcceptedBody) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + FakeResponse response(500, kPrettySuccess); + handler.OnResponse(response); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); +} + +// --------------------------------------------------------------------------- +// Asynchronous path. +// +// Export() hands the body to AsyncResponseHandler and returns before any +// response arrives, so the parsed result decides nothing the caller can see. +// It decides which internal log line is written, which is what these cases +// read. The fake delivers the response from inside SendRequest(), so the line +// is already written by the time Export() returns. +// --------------------------------------------------------------------------- +namespace +{ +struct CapturingLogHandler : public opentelemetry::sdk::common::internal_log::LogHandler +{ + void Handle(opentelemetry::sdk::common::internal_log::LogLevel level, + const char * /* file */, + int /* line */, + const char *msg, + const opentelemetry::sdk::common::AttributeMap & /* attributes */) noexcept override + { + messages.emplace_back(level, std::string{msg}); + } + + std::vector> messages; +}; + +class ElasticsearchLogsExporterAsyncTests : public ::testing::Test +{ +protected: + void SetUp() override + { + // One skip point: GTEST_SKIP returns, and a second below it would leave the rest of this body + // unreachable, which MSVC reports as C4702 under maintainer mode. +#if !defined(ENABLE_ASYNC_EXPORT) + GTEST_SKIP() << "Export() takes the synchronous path when async export is disabled"; +#elif OTEL_INTERNAL_LOG_LEVEL < OTEL_INTERNAL_LOG_LEVEL_ERROR + // These cases read the failure out of the log. Below error level it is not written at all, so + // a wrong result and a right one look the same and the cases would stop discriminating. + GTEST_SKIP() << "the failure these observe is compiled out below error level"; +#else + previous_handler_ = opentelemetry::sdk::common::internal_log::GlobalLogHandler::GetLogHandler(); + handler_ = nostd::shared_ptr( + new CapturingLogHandler()); + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogHandler(handler_); + + // The level as well as the handler. It is process wide and another case in this binary can + // lower it, and a failure filtered out at runtime looks exactly like one the parser never + // reported, so these cases would stop discriminating without saying so. + previous_level_ = opentelemetry::sdk::common::internal_log::GlobalLogHandler::GetLogLevel(); + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogLevel( + opentelemetry::sdk::common::internal_log::LogLevel::Error); + restore_level_ = true; +#endif + } + + void TearDown() override + { + // Put back what was found, rather than a fresh default that would displace a handler another + // case in this binary had installed. + if (restore_level_) + { + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogLevel(previous_level_); + } + if (handler_) + { + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogHandler(previous_handler_); + } + } + + bool LoggedAnExportFailure() const + { + const auto &messages = static_cast(handler_.get())->messages; + for (const auto &message : messages) + { + if (message.second.find("Logs were not written to Elasticsearch correctly") != + std::string::npos) + { + return true; + } + } + return false; + } + + nostd::shared_ptr handler_; + nostd::shared_ptr previous_handler_; + opentelemetry::sdk::common::internal_log::LogLevel previous_level_ = + opentelemetry::sdk::common::internal_log::LogLevel::Warning; + bool restore_level_ = false; +}; + +class ElasticsearchLogsExporterDebugLoggingTests : public ::testing::Test +{ +protected: + void SetUp() override + { + // One skip point: GTEST_SKIP returns, and a second below it would leave the rest of this body + // unreachable, which MSVC reports as C4702 under maintainer mode. +#ifdef ENABLE_ASYNC_EXPORT + GTEST_SKIP() << "Export() returns without waiting when async export is enabled"; +#elif OTEL_INTERNAL_LOG_LEVEL < OTEL_INTERNAL_LOG_LEVEL_DEBUG + // The line these read is not filtered below debug level, it is not written, so a response + // that described itself and one that stayed quiet would look the same here. + GTEST_SKIP() << "the line these observe is compiled out below debug level"; +#else + previous_handler_ = opentelemetry::sdk::common::internal_log::GlobalLogHandler::GetLogHandler(); + handler_ = nostd::shared_ptr( + new CapturingLogHandler()); + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogHandler(handler_); + + // The compiled in level only decides whether the call exists. What it dispatches is filtered + // again at runtime, and the default there is Warning, so without this the handler is installed + // and never told anything. + previous_level_ = opentelemetry::sdk::common::internal_log::GlobalLogHandler::GetLogLevel(); + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogLevel( + opentelemetry::sdk::common::internal_log::LogLevel::Debug); + restore_level_ = true; +#endif + } + + void TearDown() override + { + if (restore_level_) + { + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogLevel(previous_level_); + } + if (handler_) + { + opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogHandler(previous_handler_); + } + } + + std::size_t ResponsesDescribed() const + { + std::size_t described = 0; + for (const auto &message : static_cast(handler_.get())->messages) + { + if (message.second.find("Got response from Elasticsearch") != std::string::npos) + { + ++described; + } + } + return described; + } + + nostd::shared_ptr handler_; + nostd::shared_ptr previous_handler_; + opentelemetry::sdk::common::internal_log::LogLevel previous_level_ = + opentelemetry::sdk::common::internal_log::LogLevel::Warning; + bool restore_level_ = false; +}; +} // namespace + +// Only the response the caller was given describes itself. A second one arriving for the same +// request has its status and body dropped, so a line for it would describe an answer nobody was +// handed, and the two lines together would disagree about what the export returned. +TEST_F(ElasticsearchLogsExporterDebugLoggingTests, OnlyTheResponseThatDecidedTheOutcomeIsDescribed) +{ + const auto result = ExportWith( + [](http_client::EventHandler &handler) { + FakeResponse first(200, kCompactSuccess); + handler.OnResponse(first); + FakeResponse second(500, kCompactSuccess); + handler.OnResponse(second); + }, + 1, true); + + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); + EXPECT_EQ(ResponsesDescribed(), static_cast(1)) + << "the losing response described itself as well"; +} + +TEST_F(ElasticsearchLogsExporterAsyncTests, AnAcceptedResponseIsNotReportedAsAFailure) +{ + EXPECT_EQ(ExportWith([](http_client::EventHandler &handler) { + FakeResponse response(200, kPrettySuccess); + handler.OnResponse(response); + }), + opentelemetry::sdk::common::ExportResult::kSuccess); + + EXPECT_FALSE(LoggedAnExportFailure()); +} + +// The count reaches the parser on this path through the handler rather than through Export(), so +// a handler that dropped it would still satisfy every synchronous case. +// The status has to travel on this path too. Every other case here answers 200, so a status fixed +// at 200 would satisfy them all, and the original defect was this path ignoring the status. +TEST_F(ElasticsearchLogsExporterAsyncTests, AServerErrorIsReportedEvenWithAnAcceptedBody) +{ + EXPECT_EQ(ExportWith([](http_client::EventHandler &handler) { + FakeResponse response(500, kCompactSuccess); + handler.OnResponse(response); + }), + opentelemetry::sdk::common::ExportResult::kSuccess); + + EXPECT_TRUE(LoggedAnExportFailure()) << "a server error was not reported on the async path"; +} + +// And the count, for the same reason it is held on the synchronous path. +TEST_F(ElasticsearchLogsExporterAsyncTests, TwoAcceptedRecordsAreNotReportedAsAFailure) +{ + EXPECT_EQ(ExportWith( + [](http_client::EventHandler &handler) { + FakeResponse response(200, kTwoItemCompactSuccess); + handler.OnResponse(response); + }, + 2), + opentelemetry::sdk::common::ExportResult::kSuccess); + + EXPECT_FALSE(LoggedAnExportFailure()) << "two acknowledged records were reported as a failure"; +} + +TEST_F(ElasticsearchLogsExporterAsyncTests, ARejectedResponseIsReportedAsAFailure) +{ + // Export() reports success on this path whatever the response says, which is why the outcome has + // to be read from the log below. Asserting it here also keeps this case from having no assertion + // at all where the internal log level excludes the one that follows. + EXPECT_EQ(ExportWith( + [](http_client::EventHandler &handler) { + FakeResponse response(200, kOneItemRejected); + handler.OnResponse(response); + }, + 2), + opentelemetry::sdk::common::ExportResult::kSuccess); + + EXPECT_TRUE(LoggedAnExportFailure()); +} From d08ae971af3ac2a23a3d8e898d3c67545aa72e15 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Tue, 8 Sep 2026 06:01:45 +0000 Subject: [PATCH 2/2] [TEST] One set of fake HTTP classes, and a default body this parser accepts Main grew its own FakeResponse, FakeRequest, FakeSession and FakeHttpClient in #4071 and #4501, in an unnamed namespace, and this branch already had four of the same names in another unnamed namespace in the same file. Reopening an unnamed namespace names the same namespace, so the rebase merged both without a conflict into four redefinitions that do not compile. Merged rather than renamed. Main's set is used once and answers with one fixed response; this branch's runs a script the case supplies, which is the general case. So the session and the client take a script and default to answering the way main's did, and the existing call site is untouched. The default body had to change too, which is not cosmetic. It read `{"errors": false, "failed" : 0}`, and this pull request stops reading `failed` and starts requiring one `items` result per submitted record, so that body is now rejected: with it, ExportingARecordWithInvalidUtf8DoesNotAbort returns kFailure and fails. The default is now a body a server could send for a one record batch, and the comment that described the old substring search is rewritten. Verified in both configurations, since the guards cut different code: 26 tests, 22 passed and 4 skipped without async export, 20 passed and 6 skipped with it, no failures in either. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../test/es_log_record_exporter_test.cc | 131 +++++------------- 1 file changed, 37 insertions(+), 94 deletions(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 843340ecd2..c20759d6f9 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -34,17 +34,19 @@ namespace { namespace http_client = opentelemetry::ext::http::client; -// A response shaped like a successful Elasticsearch bulk reply: the exporter looks for -// `"failed" : 0` in the body (see ElasticsearchLogRecordExporter::Export) in addition to the -// status code before reporting success. +// What the exporter accepts: a 2xx status, "errors" false, and one item result per record the +// request submitted. A case that is not testing the response itself takes the default. +constexpr const char *kOneAcceptedRecord = + R"({"took":30,"errors":false,"items":[{"index":{"_index":"logs","_id":"1",)" + R"("_shards":{"total":2,"successful":1,"failed":0},"status":201}}]})"; + class FakeResponse final : public http_client::Response { public: - FakeResponse() - { - static const std::string kSuccessBody = R"({"errors": false, "failed" : 0})"; - body_.assign(kSuccessBody.begin(), kSuccessBody.end()); - } + explicit FakeResponse(http_client::StatusCode status = 200, + const std::string &body = kOneAcceptedRecord) + : status_(status), body_(body.begin(), body.end()) + {} const http_client::Body &GetBody() const noexcept override { return body_; } @@ -63,9 +65,10 @@ class FakeResponse final : public http_client::Response return true; } - http_client::StatusCode GetStatusCode() const noexcept override { return 200; } + http_client::StatusCode GetStatusCode() const noexcept override { return status_; } private: + http_client::StatusCode status_; http_client::Body body_; }; @@ -90,11 +93,24 @@ class FakeRequest final : public http_client::Request void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {} }; -// A session whose SendRequest() answers synchronously with a successful FakeResponse, so the -// exporter's own wait for a response returns immediately without needing a real connection. +// What the client does with a request, called from inside SendRequest() so the exporter's own +// wait returns without needing a connection. The default answers once, successfully, which is +// what a case wants when the response is not the thing under test. +using EventScript = std::function; + +EventScript AnswerSuccessfully() +{ + return [](http_client::EventHandler &handler) { + FakeResponse response; + handler.OnResponse(response); + }; +} + class FakeSession final : public http_client::Session { public: + explicit FakeSession(EventScript script = AnswerSuccessfully()) : script_(std::move(script)) {} + std::shared_ptr CreateRequest() noexcept override { return std::make_shared(); @@ -102,27 +118,34 @@ class FakeSession final : public http_client::Session void SendRequest(std::shared_ptr handler) noexcept override { - FakeResponse response; - handler->OnResponse(response); + script_(*handler); } bool IsSessionActive() noexcept override { return true; } bool CancelSession() noexcept override { return true; } bool FinishSession() noexcept override { return true; } + +private: + EventScript script_; }; class FakeHttpClient final : public http_client::HttpClient { public: + explicit FakeHttpClient(EventScript script = AnswerSuccessfully()) : script_(std::move(script)) {} + std::shared_ptr CreateSession( opentelemetry::nostd::string_view) noexcept override { - return std::make_shared(); + return std::make_shared(script_); } bool CancelAllSessions() noexcept override { return true; } bool FinishAllSessions() noexcept override { return true; } void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + EventScript script_; }; } // namespace @@ -554,86 +577,6 @@ TEST(ElasticsearchBulkResponseTests, RejectsAnItemErrorUnderErrorsFalse) // --------------------------------------------------------------------------- namespace { -namespace http_client = opentelemetry::ext::http::client; - -class FakeResponse : public http_client::Response -{ -public: - FakeResponse(http_client::StatusCode status, const std::string &body) - : status_(status), body_(body.begin(), body.end()) - {} - const http_client::Body &GetBody() const noexcept override { return body_; } - bool ForEachHeader( - nostd::function_ref) const noexcept override - { - return true; - } - bool ForEachHeader( - const nostd::string_view &, - nostd::function_ref) const noexcept override - { - return true; - } - http_client::StatusCode GetStatusCode() const noexcept override { return status_; } - -private: - http_client::StatusCode status_; - http_client::Body body_; -}; - -class FakeRequest : public http_client::Request -{ -public: - void SetMethod(http_client::Method) noexcept override {} - void SetUri(nostd::string_view) noexcept override {} - void SetSslOptions(const http_client::HttpSslOptions &) noexcept override {} - void SetBody(http_client::Body &) noexcept override {} - void AddHeader(nostd::string_view, nostd::string_view) noexcept override {} - void ReplaceHeader(nostd::string_view, nostd::string_view) noexcept override {} - void SetTimeoutMs(std::chrono::milliseconds) noexcept override {} - void SetCompression(const http_client::Compression &) noexcept override {} - void EnableLogging(bool) noexcept override {} - void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {} -}; - -using EventScript = std::function; - -class FakeSession : public http_client::Session -{ -public: - explicit FakeSession(EventScript script) : script_(std::move(script)) {} - std::shared_ptr CreateRequest() noexcept override - { - return std::make_shared(); - } - void SendRequest(std::shared_ptr handler) noexcept override - { - script_(*handler); - } - bool IsSessionActive() noexcept override { return false; } - bool CancelSession() noexcept override { return true; } - bool FinishSession() noexcept override { return true; } - -private: - EventScript script_; -}; - -class FakeHttpClient : public http_client::HttpClient -{ -public: - explicit FakeHttpClient(EventScript script) : script_(std::move(script)) {} - std::shared_ptr CreateSession(nostd::string_view) noexcept override - { - return std::make_shared(script_); - } - bool CancelAllSessions() noexcept override { return true; } - bool FinishAllSessions() noexcept override { return true; } - void SetMaxSessionsPerConnection(std::size_t) noexcept override {} - -private: - EventScript script_; -}; - // The response has to answer one operation per record, so a case using a body with N items has to // export N records or it would be rejected on the count before reaching what it means to test. opentelemetry::sdk::common::ExportResult ExportWith(EventScript script,