diff --git a/CHANGELOG.md b/CHANGELOG.md index fc64aaef23..f578920c2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -150,6 +150,14 @@ 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] Stop reading an Elasticsearch bulk response that only says the request + was accepted as a batch that was written + [#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 cfd4ce01db..999b89cdef 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(OPENTELEMETRY_INSTALL) @@ -55,4 +61,10 @@ if(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() # 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..56d0041675 --- /dev/null +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/detail/es_bulk_response.h @@ -0,0 +1,236 @@ +// 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; +} + +/** + * What an acknowledged index operation has to carry. Elasticsearch answers one with 201 when it + * created the document and 200 when it replaced an existing one. + * + * Two overloads rather than one signed parameter, because an operation status arrives through + * nlohmann::json and is compared in the type it was parsed as: is_number_integer() is true for + * unsigned as well, and 2^32 + 200 narrows back to 200 on a 32 bit int. + */ +inline bool IsAppliedStatus(nlohmann::json::number_unsigned_t value) noexcept +{ + return 200U == value || 201U == value; +} + +inline bool IsAppliedStatus(nlohmann::json::number_integer_t value) noexcept +{ + return 200 == value || 201 == value; +} + +/** + * Whether an acknowledged operation status is one that applied the operation. + * + * The type check is here rather than left to the caller, although the one caller does it too. + * This is noexcept, and get() on a value that is not a number throws, so a caller that forgot + * would not get a wrong answer, it would get std::terminate. A float does not throw: it + * truncates, so 200.5 would answer for 200. + */ +inline bool IsAcknowledgedStatus(const nlohmann::json &status) noexcept +{ + if (!status.is_number_integer()) + { + return false; + } + + if (status.is_number_unsigned()) + { + return IsAppliedStatus(status.get()); + } + + return IsAppliedStatus(status.get()); +} + +/** + * 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 af819c8eb7..4d1ff77cf1 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 @@ -458,7 +478,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 @@ -485,12 +505,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 a65c0b4c1c..21a0593991 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -3,12 +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" @@ -19,8 +24,10 @@ #include #include #include +#include #include #include +#include #include "nlohmann/json.hpp" namespace sdklogs = opentelemetry::sdk::logs; @@ -142,3 +149,704 @@ 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. +// +// This fixture is the same one #4331 adds to this file. Whichever lands first, +// the other drops the duplicate when it rebases. +// --------------------------------------------------------------------------- +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()); +}