From f9817df03512371695e3dd2448962784827ff5ee Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Mon, 31 Aug 2026 15:06:16 +0000 Subject: [PATCH] [BUG] Report an Elasticsearch async export's outcome exactly once AsyncResponseHandler called the result callback directly from OnResponse and from each terminal OnEvent state with no guard, and ReadError, WriteError and Destroyed fell through a default label and called nothing. The HTTP client can deliver both a response and a terminal event for one request, so one export could report twice, and it can end on one of those three states and report nothing at all. The exporter counts one finished session per export. Reporting twice overshoots that count for the life of the exporter. Reporting never leaves a flush waiting on a session that has already ended. Every path goes through one CompleteOnce now, a compare exchange that reports at most once and keeps the first verdict. The switch lists every state with no default, so a state added upstream fails to compile rather than going uncounted, and the destructor reports a failure for a handler torn down without an outcome. The completion line said trace span(s) in the log exporter and says log record(s) now. The cases read that line to count outcomes, and the wording was wrong either way. Nine cases drive a fake HTTP client through the public constructor: each terminal ordering a real session can produce, a response and a teardown event in both orders, and the concurrent version of each. Removing the compare exchange turns six of the nine red. Extracted from #4337, which is 1526 lines and closes two issues. What stays there is the ForceFlush deadline and watermark accounting for #4336, including four completion cases that verify this guard through the flush rather than through the log line. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 3 + .../src/es_log_record_exporter.cc | 113 ++-- .../test/es_log_record_exporter_test.cc | 564 +++++++++++++++++- 3 files changed, 645 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 821e2fb2b2..68453915d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,9 @@ Increment the: ## [Unreleased] +* [BUG] Elasticsearch: report an asynchronous export's outcome exactly once + ([#4502](https://github.com/open-telemetry/opentelemetry-cpp/pull/4502)) + * [CONFIGURATION] Apply general `attribute_limits` per individual limit field. If a model-specific limit is set it is used, otherwise the matching general limit, otherwise the model-specific default. Limit fields on diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index af819c8eb7..9079edc4af 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -267,7 +267,30 @@ class AsyncResponseHandler : public http_client::EventHandler /** * Cleans up the session in the destructor. */ - ~AsyncResponseHandler() override { session_->FinishSession(); } + ~AsyncResponseHandler() override + { + // An outcome is owed even here, or a waiter is left on a session that cannot finish. + // Reported before FinishSession(), which can block. + CompleteOnce(sdk::common::ExportResult::kFailure); + session_->FinishSession(); + } + + /** + * Report the outcome of this export, at most once. The HTTP client can deliver both a response + * and a terminal session event for one request, and the exporter counts one finished session + * per export, so only the first outcome is reported. + * @return whether this call is the one that reported. + */ + bool CompleteOnce(sdk::common::ExportResult result) noexcept + { + bool expected = false; + if (!completed_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) + { + return false; + } + result_callback_(result); + return true; + } /** * Automatically called when the response is received @@ -275,66 +298,88 @@ 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()); + const std::string body(response.GetBody().begin(), response.GetBody().end()); + const bool written = body.find("\"failed\" : 0") != std::string::npos; + + // Reported before anything is logged. CompleteOnce() retires the session and wakes + // ForceFlush() before it returns, and the log handler is replaceable, so one that calls + // ForceFlush() would otherwise wait for the session this call has not let go of. A response + // that loses the exchange says nothing either, since the outcome it would describe is not the + // one the caller was given. + if (!CompleteOnce(written ? sdk::common::ExportResult::kSuccess + : sdk::common::ExportResult::kFailure)) + { + return; + } + 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) + if (!written) { OTEL_INTERNAL_LOG_ERROR( "[ES Log Exporter] Logs were not written to Elasticsearch correctly, response body: " - << body_); - result_callback_(sdk::common::ExportResult::kFailure); - } - else - { - result_callback_(sdk::common::ExportResult::kSuccess); + << body); } } // Callback method when an http event occurs void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override { - bool need_stop = false; + // No default label, so -Wswitch reports a state added upstream rather than leaving it + // uncounted. + const char *failure = nullptr; switch (state) { + // On the way to an outcome, so nothing to report and, in particular, nothing to log: the + // session is still registered, and a replaceable log handler that flushed from here would + // wait on the export whose call stack it is standing in. + case http_client::SessionState::Created: + case http_client::SessionState::Connecting: + case http_client::SessionState::Connected: + case http_client::SessionState::Sending: + // The body arrives through OnResponse(), which is what reports the outcome. + case http_client::SessionState::Response: + break; case http_client::SessionState::CreateFailed: - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Create request to elasticsearch failed"); - need_stop = true; + failure = "[ES Log Exporter] Create request to elasticsearch failed"; + break; + case http_client::SessionState::Destroyed: + failure = "[ES Log Exporter] Session to elasticsearch destroyed before a response"; break; case http_client::SessionState::ConnectFailed: - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Connection to elasticsearch failed"); - need_stop = true; + failure = "[ES Log Exporter] Connection to elasticsearch failed"; break; case http_client::SessionState::SendFailed: - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request failed to be sent to elasticsearch"); - need_stop = true; + failure = "[ES Log Exporter] Request failed to be sent to elasticsearch"; break; case http_client::SessionState::SSLHandshakeFailed: - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] SSL handshake to elasticsearch failed"); - need_stop = true; + failure = "[ES Log Exporter] SSL handshake to elasticsearch failed"; break; case http_client::SessionState::TimedOut: - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request to elasticsearch timed out"); - need_stop = true; + failure = "[ES Log Exporter] Request to elasticsearch timed out"; break; case http_client::SessionState::NetworkError: - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Network error to elasticsearch"); - need_stop = true; + failure = "[ES Log Exporter] Network error to elasticsearch"; break; - case http_client::SessionState::Cancelled: - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request to elasticsearch cancelled"); - need_stop = true; + case http_client::SessionState::ReadError: + failure = "[ES Log Exporter] Read error"; + break; + case http_client::SessionState::WriteError: + failure = "[ES Log Exporter] Write error"; break; - default: + case http_client::SessionState::Cancelled: + failure = "[ES Log Exporter] Request to elasticsearch cancelled"; break; } - if (need_stop) + + // Logged only when this event is the outcome. These can arrive after a response, and an + // error line there would describe a failure the caller was never told about. + if (failure != nullptr && CompleteOnce(sdk::common::ExportResult::kFailure)) { - result_callback_(sdk::common::ExportResult::kFailure); + OTEL_INTERNAL_LOG_ERROR(failure); } } @@ -344,8 +389,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_ = ""; + // Whether the outcome has already been reported + std::atomic completed_{false}; // Whether to print the results from the callback bool console_debug_ = false; @@ -446,12 +491,12 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( { OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export " << span_count - << " trace span(s) error: " << static_cast(result)); + << " log record(s) error: " << static_cast(result)); } else { OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count - << " trace span(s) success"); + << " log record(s) success"); } synchronization_data->finished_session_counter_.fetch_add(1, std::memory_order_release); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index a65c0b4c1c..b6758b8374 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -4,11 +4,14 @@ #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h" #include "opentelemetry/common/timestamp.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/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" @@ -16,17 +19,27 @@ #include #include -#include +#include #include #include +#include +#include #include +#include #include +#include +// nlohmann is used through its public header only, which is what every other file here +// does. The detail headers below do not exist when it is installed as one amalgamated +// header, so asking for them breaks that build. +// IWYU pragma: no_include +// IWYU pragma: no_include #include "nlohmann/json.hpp" namespace sdklogs = opentelemetry::sdk::logs; namespace logs_api = opentelemetry::logs; namespace nostd = opentelemetry::nostd; namespace logs_exporter = opentelemetry::exporter::logs; +namespace internal_log = opentelemetry::sdk::common::internal_log; TEST(ElasticsearchLogsExporterTests, CustomClientConstructionSucceeds) { @@ -142,3 +155,552 @@ TEST(ElasticsearchLogRecordableTests, BasicTests) EXPECT_EQ(actual, expected); } + +// --------------------------------------------------------------------------- +// ForceFlush deadline. +// --------------------------------------------------------------------------- +namespace +{ +namespace http_client = opentelemetry::ext::http::client; + +// Accepted by the substring check, by a top level "errors": false parse, and by one +// acknowledged operation result carrying a 2xx status, so these cases keep meaning the +// same thing whichever success check is in place. +constexpr const char *kAcceptedBody = + R"({"took":30,"errors":false,"items":[{"index":{"status":201,"_shards":{"failed" : 0}}}]})"; + +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 + { + if (on_create_session) + { + on_create_session(); + } + return std::make_shared(script_); + } + + // Runs inside Export(), after the records have been handed over and before the request exists. + std::function on_create_session; + + // Runs inside Shutdown(). A real client answers its outstanding sessions here, so a case that + // needs a flush to be woken by the shutdown rather than by its own bound sets this; one that + // leaves it unset is a client that goes quiet instead, which is the case the bound exists for. + std::function on_cancel_all; + + bool CancelAllSessions() noexcept override + { + if (on_cancel_all) + { + on_cancel_all(); + } + return true; + } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + EventScript script_; +}; + +} // namespace + +// --------------------------------------------------------------------------- +// A fake HTTP client, and an exporter built on it, shared by the cases below. +// --------------------------------------------------------------------------- +namespace +{ +// A response timeout short enough that a wait bounded by it instead of by the caller's deadline +// is visible in the elapsed time. +constexpr int kShortResponseTimeoutSeconds = 2; + +struct FlushFixture +{ + std::shared_ptr client; + std::unique_ptr exporter; +}; + +FlushFixture MakeExporter(EventScript script) +{ + FlushFixture fixture; + fixture.client = std::make_shared(std::move(script)); + logs_exporter::ElasticsearchExporterOptions options; + options.response_timeout_ = kShortResponseTimeoutSeconds; + fixture.exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, fixture.client)); + return fixture; +} + +void ExportOnce(logs_exporter::ElasticsearchLogRecordExporter &exporter) +{ + auto record = exporter.MakeRecordable(); + exporter.Export(nostd::span>(&record, 1)); +} +} // namespace + +// --------------------------------------------------------------------------- +// Exactly-once accounting for the async handler, which exists only in an async build, so +// these cases skip there rather than compile out. +// --------------------------------------------------------------------------- + +namespace +{ +// The completion callback logs one line per invocation and names the verdict in it, so these +// count the callback and say which result it carried. +// +// Session tracking cannot stand in for this: ids are erased, and erasing one that has already gone +// is a no-op, so ForceFlush() reports the same thing whether the callback ran once or three times. +class CompletionCountingLogHandler : public internal_log::LogHandler +{ +public: + void Handle(internal_log::LogLevel /* level */, + const char * /* file */, + int /* line */, + const char *msg, + const opentelemetry::sdk::common::AttributeMap & /* attributes */) noexcept override + { + if (msg == nullptr) + { + return; + } + lines_.fetch_add(1, std::memory_order_relaxed); + + const std::string text(msg); + if (text.find("log record(s) success") != std::string::npos) + { + successes_.fetch_add(1, std::memory_order_relaxed); + } + else if (text.find("log record(s) error") != std::string::npos) + { + failures_.fetch_add(1, std::memory_order_relaxed); + } + } + + int successes() const noexcept { return successes_.load(std::memory_order_relaxed); } + int failures() const noexcept { return failures_.load(std::memory_order_relaxed); } + int completions() const noexcept { return successes() + failures(); } + + // Everything the handler was given, not only the completions. What a session says on its way to + // an outcome is as much a part of the contract as what it says at the end of one. + int lines() const noexcept { return lines_.load(std::memory_order_relaxed); } + +private: + std::atomic successes_{0}; + std::atomic failures_{0}; + std::atomic lines_{0}; +}; + +class ElasticsearchAsyncCompletionTests : public ::testing::Test +{ +protected: + void SetUp() override + { +#if !defined(ENABLE_ASYNC_EXPORT) + GTEST_SKIP() << "the async handler does not exist when async export is disabled"; +#elif OTEL_INTERNAL_LOG_LEVEL < OTEL_INTERNAL_LOG_LEVEL_DEBUG + GTEST_SKIP() << "the success half of the completion callback is compiled out below debug level"; +#else + // One skip point, because GTEST_SKIP returns and a second one after it would leave the rest of + // this body unreachable, which MSVC reports as C4702 under maintainer mode. + previous_handler_ = internal_log::GlobalLogHandler::GetLogHandler(); + handler_ = nostd::shared_ptr(new CompletionCountingLogHandler()); + internal_log::GlobalLogHandler::SetLogHandler(handler_); + previous_level_ = internal_log::GlobalLogHandler::GetLogLevel(); + internal_log::GlobalLogHandler::SetLogLevel(internal_log::LogLevel::Debug); +#endif + } + + void TearDown() override + { + if (handler_) + { + internal_log::GlobalLogHandler::SetLogLevel(previous_level_); + internal_log::GlobalLogHandler::SetLogHandler(previous_handler_); + } + } + + const CompletionCountingLogHandler &Counter() const + { + return *static_cast(handler_.get()); + } + + int Completions() const { return Counter().completions(); } + int Lines() const { return Counter().lines(); } + + nostd::shared_ptr handler_; + nostd::shared_ptr previous_handler_; + internal_log::LogLevel previous_level_ = internal_log::LogLevel::Warning; +}; +} // namespace +// The orderings a real session can produce, each of which reported twice before the guard. +TEST_F(ElasticsearchAsyncCompletionTests, TerminalOrderingsReportExactlyOnce) +{ + using State = http_client::SessionState; + struct Case + { + const char *name; + State first; + State second; + }; + const Case cases[] = { + {"connect then create", State::ConnectFailed, State::CreateFailed}, + {"read error then destroyed", State::ReadError, State::Destroyed}, + {"write error then destroyed", State::WriteError, State::Destroyed}, + {"timed out then network error", State::TimedOut, State::NetworkError}, + {"cancelled then destroyed", State::Cancelled, State::Destroyed}, + }; + + for (const auto &test_case : cases) + { + SCOPED_TRACE(test_case.name); + std::vector> kept; + auto fixture = MakeExporter( + [&kept, &test_case](const std::shared_ptr &handler) { + kept.push_back(handler); + handler->OnEvent(test_case.first, ""); + handler->OnEvent(test_case.second, ""); + }); + + const int before = Completions(); + ExportOnce(*fixture.exporter); + EXPECT_EQ(Completions() - before, 1); + + // The handler is still alive at the check above, and its destructor reports when nothing + // else has. Letting it go here is what makes the two together exactly one rather than the + // callback alone. + kept.clear(); + EXPECT_EQ(Completions() - before, 1) << "destroying the handler reported a second time"; + } +} + +// A response decides the outcome, and a teardown event arriving after it must not report again. +// The other order is the case below, because the first verdict is the one that has to survive +// either way round. +TEST_F(ElasticsearchAsyncCompletionTests, AResponseAndATeardownEventReportOnce) +{ + for (const auto state : + {http_client::SessionState::Destroyed, http_client::SessionState::Cancelled, + http_client::SessionState::TimedOut}) + { + SCOPED_TRACE(static_cast(state)); + std::vector> kept; + auto fixture = + MakeExporter([&kept, state](const std::shared_ptr &handler) { + kept.push_back(handler); + FakeResponse response(200, kAcceptedBody); + handler->OnResponse(response); + handler->OnEvent(state, ""); + }); + + const int before = Completions(); + const int failures_before = Counter().failures(); + ExportOnce(*fixture.exporter); + + EXPECT_EQ(Completions() - before, 1); + EXPECT_EQ(Counter().failures() - failures_before, 0) + << "the teardown verdict replaced the response's"; + + kept.clear(); + EXPECT_EQ(Completions() - before, 1) << "destroying the handler reported a second time"; + } +} + +// The other order, and the contract it settles. A read or write error ends the export here: the +// exporter treats it as the outcome, and a response arriving afterwards is ignored rather than +// replacing it. EventHandler does not say whether either state can be followed by a response, so +// this is the choice this exporter makes, written down where a change to it would be visible. +TEST_F(ElasticsearchAsyncCompletionTests, ATeardownEventAndALaterResponseReportOnce) +{ + for (const auto state : + {http_client::SessionState::ReadError, http_client::SessionState::WriteError, + http_client::SessionState::Destroyed, http_client::SessionState::TimedOut, + http_client::SessionState::NetworkError, http_client::SessionState::Cancelled}) + { + SCOPED_TRACE(static_cast(state)); + std::vector> kept; + auto fixture = + MakeExporter([&kept, state](const std::shared_ptr &handler) { + kept.push_back(handler); + handler->OnEvent(state, ""); + FakeResponse response(200, kAcceptedBody); + handler->OnResponse(response); + }); + + const int before = Completions(); + const int failures_before = Counter().failures(); + ExportOnce(*fixture.exporter); + + EXPECT_EQ(Completions() - before, 1); + EXPECT_EQ(Counter().failures() - failures_before, 1) + << "a response after the failure replaced the verdict that had already been reported"; + + kept.clear(); + EXPECT_EQ(Completions() - before, 1) << "destroying the handler reported a second time"; + } +} + +// Two terminal events delivered at the same time. The inline scripts above cannot reach the race +// the compare-exchange exists for. +TEST_F(ElasticsearchAsyncCompletionTests, ConcurrentTerminalEventsReportOnce) +{ + std::shared_ptr captured; + auto fixture = + MakeExporter([&captured](const std::shared_ptr &handler) { + captured = handler; + }); + ExportOnce(*fixture.exporter); + ASSERT_NE(captured, nullptr); + + std::atomic go{false}; + std::thread first([&captured, &go] { + while (!go.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + captured->OnEvent(http_client::SessionState::ConnectFailed, ""); + }); + std::thread second([&captured, &go] { + while (!go.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + captured->OnEvent(http_client::SessionState::NetworkError, ""); + }); + go.store(true, std::memory_order_release); + first.join(); + second.join(); + + EXPECT_EQ(Completions(), 1); +} + +// A response and a terminal event delivered at the same time. Whichever wins, there is one report. +TEST_F(ElasticsearchAsyncCompletionTests, AConcurrentResponseAndTerminalEventReportOnce) +{ + std::shared_ptr captured; + auto fixture = + MakeExporter([&captured](const std::shared_ptr &handler) { + captured = handler; + }); + ExportOnce(*fixture.exporter); + ASSERT_NE(captured, nullptr); + + std::atomic go{false}; + std::thread responder([&captured, &go] { + while (!go.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + FakeResponse response(200, kAcceptedBody); + captured->OnResponse(response); + }); + std::thread failer([&captured, &go] { + while (!go.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + captured->OnEvent(http_client::SessionState::TimedOut, ""); + }); + go.store(true, std::memory_order_release); + responder.join(); + failer.join(); + + EXPECT_EQ(Completions(), 1); +} + +// What Session::SendRequest does when HttpOperation::SendAsync fails to set up: the operation +// dispatches ConnectFailed and returns non-OK, then SendRequest dispatches CreateFailed for the +// same handler. One export, so one finished session, not two. +namespace +{ +// Calls back into the exporter from inside the log handler, which is what an application can +// install through GlobalLogHandler::SetLogHandler(). +class FlushingLogHandler : public internal_log::LogHandler +{ +public: + // The needle picks which diagnostic re-enters the exporter, because the two paths that log + // one describe it differently. + void Watch(logs_exporter::ElasticsearchLogRecordExporter *exporter, + const char *needle = "Logs were not written") noexcept + { + exporter_ = exporter; + needle_ = needle; + } + + void Handle(internal_log::LogLevel /* level */, + const char * /* file */, + int /* line */, + const char *msg, + const opentelemetry::sdk::common::AttributeMap & /* attributes */) noexcept override + { + if (msg == nullptr || exporter_ == nullptr) + { + return; + } + if (std::string(msg).find(needle_) == std::string::npos) + { + return; + } + lines_.fetch_add(1, std::memory_order_relaxed); + if (reentered_.exchange(true, std::memory_order_relaxed)) + { + return; + } + flushed_.store(exporter_->ForceFlush(std::chrono::milliseconds{20}), std::memory_order_relaxed); + } + + bool reentered() const noexcept { return reentered_.load(std::memory_order_relaxed); } + bool flushed() const noexcept { return flushed_.load(std::memory_order_relaxed); } + int lines() const noexcept { return lines_.load(std::memory_order_relaxed); } + +private: + logs_exporter::ElasticsearchLogRecordExporter *exporter_{nullptr}; + const char *needle_{"Logs were not written"}; + std::atomic reentered_{false}; + std::atomic flushed_{false}; + std::atomic lines_{0}; +}; +} // namespace + +// The session has to be retired before anything replaceable is called, or a handler that flushes +// waits for the export whose completion is calling it. +TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromInsideTheLogHandlerDoesNotWaitForItsOwnSession) +{ + auto fixture = MakeExporter([](const std::shared_ptr &handler) { + FakeResponse response(200, R"({"took":1,"errors":true,"items":[]})"); + handler->OnResponse(response); + }); + + auto watcher = nostd::shared_ptr(new FlushingLogHandler()); + auto *raw = static_cast(watcher.get()); + raw->Watch(fixture.exporter.get()); + internal_log::GlobalLogHandler::SetLogHandler(watcher); + + ExportOnce(*fixture.exporter); + + ASSERT_TRUE(raw->reentered()) << "the failure never reached the log handler"; + EXPECT_TRUE(raw->flushed()) << "the flush waited for the session that was reporting itself"; + raw->Watch(nullptr); +} + +// The same rule on the path that refuses the batch. The export is registered before the shutdown +// check, so reporting the refusal before retiring it makes a flushing handler wait for the +// Export() that is calling it, and the refusal is described twice. +TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromTheShutdownErrorDoesNotWaitForItsOwnExport) +{ + auto fixture = MakeExporter([](const std::shared_ptr &) {}); + ASSERT_TRUE(fixture.exporter->Shutdown()); + + auto watcher = nostd::shared_ptr(new FlushingLogHandler()); + auto *raw = static_cast(watcher.get()); + raw->Watch(fixture.exporter.get(), "exporter is shutdown"); + internal_log::GlobalLogHandler::SetLogHandler(watcher); + + ExportOnce(*fixture.exporter); + + ASSERT_TRUE(raw->reentered()) << "the shutdown refusal never reached the log handler"; + EXPECT_TRUE(raw->flushed()) << "the flush waited for the export that was refusing itself"; + EXPECT_EQ(1, raw->lines()) << "one refusal was described " << raw->lines() << " times"; + raw->Watch(nullptr); +} + +// Two responses for one request write the same body and race for the same outcome. The body is a +// local so there is nothing shared to tear, and the exchange decides which one reports. +TEST_F(ElasticsearchAsyncCompletionTests, TwoConcurrentResponsesReportOnce) +{ + std::shared_ptr captured; + auto fixture = + MakeExporter([&captured](const std::shared_ptr &handler) { + captured = handler; + }); + ExportOnce(*fixture.exporter); + ASSERT_NE(captured, nullptr); + + std::atomic go{false}; + const auto deliver = [&captured, &go](const char *body) { + while (!go.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + FakeResponse response(200, body); + captured->OnResponse(response); + }; + std::thread first(deliver, kAcceptedBody); + std::thread second(deliver, R"({"took":2,"errors":true,"items":[]})"); + go.store(true, std::memory_order_release); + first.join(); + second.join(); + + EXPECT_EQ(Completions(), 1) << "one request, one outcome, whichever response won"; +} + +// A handler destroyed without ever reporting still has to finish its session. +TEST_F(ElasticsearchAsyncCompletionTests, AHandlerDestroyedWithoutAnOutcomeStillFinishes) +{ + auto fixture = MakeExporter([](const std::shared_ptr &) {}); + ExportOnce(*fixture.exporter); + EXPECT_TRUE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); +}