diff --git a/CHANGELOG.md b/CHANGELOG.md index 098a96cc33..dcc43bc6c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,18 @@ Increment the: ## [Unreleased] +* [BUG] Elasticsearch: report an asynchronous export's outcome exactly once + [#4502](https://github.com/open-telemetry/opentelemetry-cpp/pull/4502) + +* [BUG] Elasticsearch: retire an asynchronous session before the completion + callback writes its diagnostic, so a log handler that flushes does not wait + for the export reporting to it + [#4502](https://github.com/open-telemetry/opentelemetry-cpp/pull/4502) + +* [CODE HEALTH] Elasticsearch: say log record(s) rather than trace span(s) in + the asynchronous export result + [#4502](https://github.com/open-telemetry/opentelemetry-cpp/pull/4502) + ## [1.30.0] 2026-10-22 This release contains: diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index cd3b1bdbfd..741eb9c6c2 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -267,7 +267,28 @@ class AsyncResponseHandler : public http_client::EventHandler /** * Cleans up the session in the destructor. */ - ~AsyncResponseHandler() override { session_->FinishSession(); } + ~AsyncResponseHandler() override + { + // Report before FinishSession(), which can block: an outcome is owed even here. + CompleteOnce(sdk::common::ExportResult::kFailure); + session_->FinishSession(); + } + + /** + * Report this export's outcome, at most once: one request can deliver both a response and a + * terminal event, and the exporter counts one finished session per export. + * @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 +296,82 @@ 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; + + // Report before logging: CompleteOnce() retires the session, and a replaceable handler that + // flushes would wait on it. A loser stays silent; its outcome went to nobody. + 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: -Wswitch then reports a state added upstream instead of dropping it. + const char *failure = nullptr; switch (state) { + // Progress only. The session is still registered, so a handler that flushed from a log + // line here would wait on the export 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) + + // Only the event that decided the outcome speaks; a later one names a failure nobody got. + if (failure != nullptr && CompleteOnce(sdk::common::ExportResult::kFailure)) { - result_callback_(sdk::common::ExportResult::kFailure); + OTEL_INTERNAL_LOG_ERROR(failure); } } @@ -344,8 +381,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; @@ -448,20 +485,22 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( auto handler = std::make_shared( session, [span_count, synchronization_data](opentelemetry::sdk::common::ExportResult result) { + // Count and wake before logging: a handler that flushes from the line below would + // wait for the session reporting to it. + synchronization_data->finished_session_counter_.fetch_add(1, std::memory_order_release); + synchronization_data->force_flush_cv.notify_all(); + if (result != opentelemetry::sdk::common::ExportResult::kSuccess) { 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); - synchronization_data->force_flush_cv.notify_all(); return true; }, options_.console_debug_); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 52a5d9b6ef..58c8359dc6 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -11,6 +11,7 @@ #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" @@ -18,28 +19,36 @@ #include #include +#include #include #include #include +#include +#include #include +#include #include +#include +// nlohmann is used through its public header only, like every other file here. The detail +// headers below do not exist in the amalgamated install, so asking for them breaks that build. +// IWYU pragma: no_include +// IWYU pragma: no_include #include "nlohmann/json.hpp" 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. +// A bulk reply the exporter accepts: it looks for `"failed" : 0` as well as the status code. +constexpr const char *kDefaultAcceptedBody = R"({"errors": false, "failed" : 0})"; + 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 = kDefaultAcceptedBody) + : status_(status), body_(body.begin(), body.end()) + {} const http_client::Body &GetBody() const noexcept override { return body_; } @@ -58,9 +67,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_; }; @@ -85,11 +95,23 @@ 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, run from inside SendRequest() so no connection is +// needed. The handler is shared so a case can keep it and deliver a second event. +using EventScript = std::function &)>; + +EventScript AnswerSuccessfully() +{ + return [](const std::shared_ptr &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(); @@ -97,27 +119,52 @@ 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(); + 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(): set for a client that answers there, unset for one that stays quiet. + std::function on_cancel_all; + + bool CancelAllSessions() noexcept override + { + if (on_cancel_all) + { + on_cancel_all(); + } + return true; } - bool CancelAllSessions() noexcept override { return true; } bool FinishAllSessions() noexcept override { return true; } void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + EventScript script_; }; } // namespace @@ -126,6 +173,7 @@ 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; // Regression test: a log record whose body carries bytes that are not valid UTF-8 used to // abort the process. ElasticSearchRecordable::WriteValue stores the value as given, and @@ -263,3 +311,496 @@ TEST(ElasticsearchLogRecordableTests, BasicTests) EXPECT_EQ(actual, expected); } + +// --------------------------------------------------------------------------- +// Async completion and terminal diagnostics. +// --------------------------------------------------------------------------- +namespace +{ +namespace http_client = opentelemetry::ext::http::client; + +// A success response under either success check: a substring search wants the spacing in +// `"failed" : 0`, and parsing the operation results wants a string `_index` on each. +constexpr const char *kAcceptedBody = + R"({"took":30,"errors":false,"items":[{"index":{"_index":"logs","status":201,)" + R"("_shards":{"failed" : 0}}}]})"; + +} // 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; +// A flush with nothing outstanding returns at once and one that waits burns at least its own +// timeout, so this sits between the two whether or not the caller deadline is honoured (#4336). +constexpr auto kFlushTimeout = std::chrono::milliseconds{200}; +constexpr std::int64_t kFlushDidNotWaitUs = 100000; + +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; +} + +// Microseconds a flush took, for the cases that hold it did not wait for anything. +std::int64_t FlushUs(logs_exporter::ElasticsearchLogRecordExporter &exporter) +{ + const auto started = std::chrono::steady_clock::now(); + exporter.ForceFlush(kFlushTimeout); + return std::chrono::duration_cast(std::chrono::steady_clock::now() - + started) + .count(); +} + +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. + 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; + } + // ForceFlush() also returns true when it gives up, so only the duration tells a flush with + // nothing to wait for from one that waited out the response timeout. + flush_us_.store(FlushUs(*exporter_), std::memory_order_relaxed); + } + + bool reentered() const noexcept { return reentered_.load(std::memory_order_relaxed); } + std::int64_t flush_us() const noexcept { return flush_us_.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 flush_us_{0}; + 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_LT(raw->flush_us(), kFlushDidNotWaitUs) + << "the flush waited " << raw->flush_us() << "us for the session that was reporting itself"; + raw->Watch(nullptr); +} + +// The callback CompleteOnce() invokes writes its own diagnostic, and it is replaceable too, so +// the session has to be counted and woken before that line rather than after it. +TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromTheCompletionErrorDoesNotWaitForItsOwnSession) +{ + 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(), "ERROR: Export"); + internal_log::GlobalLogHandler::SetLogHandler(watcher); + + ExportOnce(*fixture.exporter); + + ASSERT_TRUE(raw->reentered()) << "the failure never reached the log handler"; + EXPECT_LT(raw->flush_us(), kFlushDidNotWaitUs) + << "the flush waited " << raw->flush_us() << "us for the session that was reporting itself"; + raw->Watch(nullptr); +} + +// The same property on the path that refuses the batch: a handler that flushes from inside the +// refusal must not wait for the Export() calling it. The shutdown check returns before +// session_counter_ is incremented, so nothing is ever registered for a refused export. +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_LT(raw->flush_us(), kFlushDidNotWaitUs) + << "the flush waited " << raw->flush_us() << "us 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); + + const std::int64_t waited = FlushUs(*fixture.exporter); + EXPECT_LT(waited, kFlushDidNotWaitUs) + << "the flush waited " << waited << "us for a session that reported nothing"; +}