From 0470e29d5a855e62d37916f5f9d32a0bf182bbb6 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 1/7] [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 | 562 +++++++++++++++++- 3 files changed, 643 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e203290157..6f7329240b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,9 @@ Increment the: to compile standalone on newer standard library implementations. [#4574](https://github.com/open-telemetry/opentelemetry-cpp/pull/4574) +* [BUG] Elasticsearch: report an asynchronous export's outcome exactly once + [#4502](https://github.com/open-telemetry/opentelemetry-cpp/pull/4502) + ## [1.29.0] 2026-09-13 * [RELEASE] Bump main branch to 1.29.0-dev (#4259) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index cd3b1bdbfd..596e96f962 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; @@ -452,12 +497,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 52a5d9b6ef..1b3aead501 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,11 +19,20 @@ #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 @@ -126,6 +136,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 +274,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})); +} From aa5d5f7a0e6a5424eb58a8f8585df186ec7cc298 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Tue, 8 Sep 2026 06:48:18 +0000 Subject: [PATCH 2/7] [TEST] One set of fake HTTP classes after main added its own Main's #4071 and #4501 put FakeResponse, FakeRequest, FakeSession and FakeHttpClient in an unnamed namespace at the top of this file, and this branch already had four of those names in a second unnamed namespace lower down. Reopening an unnamed namespace names the same namespace, so the rebase merged both with no conflict at all and left four redefinitions. One set now. The session and the client take a script and default to answering the way main's did, so main's own call site needs no edit. The script carries the handler as a shared_ptr rather than a reference, because the cases here have to keep it and send a second event to it, and the client keeps its on_create_session and on_cancel_all hooks, both empty by default. The default body stays as main wrote it. This branch does not change how the exporter decides success, so what main's cases send still passes here. Verified in both configurations with maintainer mode on: 12 cases, 12 passing with async export and 3 passing with 9 skipping without it. Removing CompleteOnce's compare and exchange turns six of them red, so the rewritten fixtures still discriminate. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../test/es_log_record_exporter_test.cc | 166 ++++++------------ 1 file changed, 56 insertions(+), 110 deletions(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 1b3aead501..e7a77dc0ed 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -42,14 +42,15 @@ 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. +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_; } @@ -68,9 +69,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_; }; @@ -95,11 +97,26 @@ class FakeRequest final : public http_client::Request void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {} }; -// A session whose SendRequest() answers synchronously with a successful FakeResponse, so the -// exporter's own wait for a response returns immediately without needing a real connection. +// What the client does with a request, called from inside SendRequest() so the exporter's own +// wait returns without needing a connection. The handler travels as a shared_ptr because a case +// that checks an outcome is reported once has to keep it and send a second event to it. The +// default answers once, successfully, which is what a case wants when the response is not the +// thing under test. +using EventScript = std::function &)>; + +EventScript AnswerSuccessfully() +{ + return [](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(); @@ -107,27 +124,54 @@ 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(). 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 CancelAllSessions() noexcept override { return true; } bool FinishAllSessions() noexcept override { return true; } void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + EventScript script_; }; } // namespace @@ -288,104 +332,6 @@ namespace http_client = opentelemetry::ext::http::client; 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 // --------------------------------------------------------------------------- From e2fda01299ab94d82599f126b66cb2115a1af2c0 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Mon, 21 Sep 2026 17:36:23 +0000 Subject: [PATCH 3/7] [TEST] Make the re-entrant flush cases able to fail Both asserted EXPECT_TRUE(flushed()), and ForceFlush() reports success when it gives up on its own condition variable, so the return value cannot tell a flush that had nothing to wait for from one that waited the whole response timeout. Inverting either ordering left both cases green at 2000 ms, the timeout they exist to rule out. Time the re-entrant flush and bound it instead. Also correct the comment on AFlushFromTheShutdownErrorDoesNotWaitForItsOwnExport. It said the export is registered before the shutdown check; the check returns first, so nothing is registered for a refused export, which is what makes the re-entrant flush return immediately on that path. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../test/es_log_record_exporter_test.cc | 30 ++++++++++++++----- 1 file changed, 22 insertions(+), 8 deletions(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index e7a77dc0ed..119f020cd8 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -342,6 +343,9 @@ 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 that waits for its own export blocks until that timeout, so half of it separates +// the two outcomes with a wide margin either way. +constexpr std::int64_t kFlushDidNotWaitUs = kShortResponseTimeoutSeconds * 500000; struct FlushFixture { @@ -674,18 +678,26 @@ class FlushingLogHandler : public internal_log::LogHandler { return; } - flushed_.store(exporter_->ForceFlush(std::chrono::milliseconds{20}), std::memory_order_relaxed); + // ForceFlush() reports success when it gives up on its own condition variable, so the return + // value cannot tell a flush that had nothing to wait for from one that waited the whole + // response timeout. The duration can. + const auto started = std::chrono::steady_clock::now(); + exporter_->ForceFlush(std::chrono::milliseconds{20}); + flush_us_.store(std::chrono::duration_cast( + std::chrono::steady_clock::now() - started) + .count(), + 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); } + 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 flushed_{false}; + std::atomic flush_us_{0}; std::atomic lines_{0}; }; } // namespace @@ -707,13 +719,14 @@ TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromInsideTheLogHandlerDoesNotWa 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"; + 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 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. +// 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 &) {}); @@ -727,7 +740,8 @@ TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromTheShutdownErrorDoesNotWaitF 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_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); } From 08d01e664aceda984b6f3774076ce75ee1a80617 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Mon, 21 Sep 2026 17:41:49 +0000 Subject: [PATCH 4/7] [BUG] Retire the session before the completion callback logs OnResponse() reports through CompleteOnce() before it logs, so a log handler that calls ForceFlush() does not wait for the session reporting to it. The callback CompleteOnce() invokes does the opposite: it writes its own diagnostic first and counts the session finished afterwards, so a handler watching that line waits the whole response timeout. Measured at 2000158us. Count and wake first, then log, and cover that message with a case of its own. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../src/es_log_record_exporter.cc | 9 +++++--- .../test/es_log_record_exporter_test.cc | 22 +++++++++++++++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 596e96f962..0b0d85e38c 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -493,6 +493,12 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( auto handler = std::make_shared( session, [span_count, synchronization_data](opentelemetry::sdk::common::ExportResult result) { + // Counted and woken before anything replaceable runs, for the same reason OnResponse() + // reports before it logs: a handler that calls ForceFlush() from the line below would + // otherwise 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 " @@ -504,9 +510,6 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count << " 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 119f020cd8..c612325d29 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -724,6 +724,28 @@ TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromInsideTheLogHandlerDoesNotWa 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. From fe7e109048927a599b7b4fab2e8717c3eed61d33 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Mon, 21 Sep 2026 18:02:34 +0000 Subject: [PATCH 5/7] [TEST] Bound the destructor flush case, and tighten the comments AHandlerDestroyedWithoutAnOutcomeStillFinishes asserted EXPECT_TRUE(ForceFlush(...)) like the other two did, so removing the completion from ~AsyncResponseHandler left it green at 2000 ms. It now times the flush through the same helper the re-entrant cases use. The comments the change adds are cut to describe the code rather than explain it, none longer than what it annotates. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../src/es_log_record_exporter.cc | 31 ++++------- .../test/es_log_record_exporter_test.cc | 53 +++++++++---------- 2 files changed, 36 insertions(+), 48 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 0b0d85e38c..741eb9c6c2 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -269,16 +269,14 @@ class AsyncResponseHandler : public http_client::EventHandler */ ~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. + // Report before FinishSession(), which can block: an outcome is owed even here. 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. + * 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 @@ -301,11 +299,8 @@ class AsyncResponseHandler : public http_client::EventHandler 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. + // 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)) { @@ -328,14 +323,12 @@ class AsyncResponseHandler : public http_client::EventHandler // Callback method when an http event occurs void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override { - // No default label, so -Wswitch reports a state added upstream rather than leaving it - // uncounted. + // No default label: -Wswitch then reports a state added upstream instead of dropping it. 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. + // 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: @@ -375,8 +368,7 @@ class AsyncResponseHandler : public http_client::EventHandler break; } - // 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. + // Only the event that decided the outcome speaks; a later one names a failure nobody got. if (failure != nullptr && CompleteOnce(sdk::common::ExportResult::kFailure)) { OTEL_INTERNAL_LOG_ERROR(failure); @@ -493,9 +485,8 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( auto handler = std::make_shared( session, [span_count, synchronization_data](opentelemetry::sdk::common::ExportResult result) { - // Counted and woken before anything replaceable runs, for the same reason OnResponse() - // reports before it logs: a handler that calls ForceFlush() from the line below would - // otherwise wait for the session reporting to it. + // 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(); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index c612325d29..83fbb5aa85 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -40,9 +40,7 @@ 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 @@ -98,11 +96,8 @@ class FakeRequest final : public http_client::Request void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {} }; -// What the client does with a request, called from inside SendRequest() so the exporter's own -// wait returns without needing a connection. The handler travels as a shared_ptr because a case -// that checks an outcome is reported once has to keep it and send a second event to it. The -// default answers once, successfully, which is what a case wants when the response is not the -// thing under test. +// 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() @@ -154,9 +149,7 @@ class FakeHttpClient final : public http_client::HttpClient // 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. + // 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 @@ -327,9 +320,8 @@ 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. +// Accepted by the substring check, by an "errors": false parse and by a 2xx operation result, +// so these cases mean 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}}}]})"; @@ -343,8 +335,7 @@ 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 that waits for its own export blocks until that timeout, so half of it separates -// the two outcomes with a wide margin either way. +// A flush that waits for its own export burns that timeout, so half of it separates the two. constexpr std::int64_t kFlushDidNotWaitUs = kShortResponseTimeoutSeconds * 500000; struct FlushFixture @@ -364,6 +355,16 @@ FlushFixture MakeExporter(EventScript script) 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(std::chrono::milliseconds{20}); + return std::chrono::duration_cast(std::chrono::steady_clock::now() - + started) + .count(); +} + void ExportOnce(logs_exporter::ElasticsearchLogRecordExporter &exporter) { auto record = exporter.MakeRecordable(); @@ -413,8 +414,7 @@ class CompletionCountingLogHandler : public internal_log::LogHandler 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. + // Everything the handler was given, not only the completions. int lines() const noexcept { return lines_.load(std::memory_order_relaxed); } private: @@ -678,15 +678,9 @@ class FlushingLogHandler : public internal_log::LogHandler { return; } - // ForceFlush() reports success when it gives up on its own condition variable, so the return - // value cannot tell a flush that had nothing to wait for from one that waited the whole - // response timeout. The duration can. - const auto started = std::chrono::steady_clock::now(); - exporter_->ForceFlush(std::chrono::milliseconds{20}); - flush_us_.store(std::chrono::duration_cast( - std::chrono::steady_clock::now() - started) - .count(), - std::memory_order_relaxed); + // 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); } @@ -803,5 +797,8 @@ TEST_F(ElasticsearchAsyncCompletionTests, AHandlerDestroyedWithoutAnOutcomeStill { auto fixture = MakeExporter([](const std::shared_ptr &) {}); ExportOnce(*fixture.exporter); - EXPECT_TRUE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); + + const std::int64_t waited = FlushUs(*fixture.exporter); + EXPECT_LT(waited, kFlushDidNotWaitUs) + << "the flush waited " << waited << "us for a session that reported nothing"; } From 267442ec9fe78f45b41ce82292965100eb598524 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Mon, 21 Sep 2026 18:40:40 +0000 Subject: [PATCH 6/7] [TEST] Take the flush bound from the request, not the response timeout kFlushDidNotWaitUs came from options_.response_timeout_, which is only what a waiting flush burns because ForceFlush ignores the caller deadline (#4336). A fix there would have left a broken ordering waiting 20ms, under the bound, and every one of these cases would have stopped failing without saying so. Ask for 200ms and bound at half of it, so the two outcomes stay apart either way. Checked by applying a caller-deadline fix and re-running the mutation: it fails at 200110us instead of 2000434us. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../elasticsearch/test/es_log_record_exporter_test.cc | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 83fbb5aa85..3d3b41f336 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -335,8 +335,10 @@ 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 that waits for its own export burns that timeout, so half of it separates the two. -constexpr std::int64_t kFlushDidNotWaitUs = kShortResponseTimeoutSeconds * 500000; +// 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 { @@ -359,7 +361,7 @@ FlushFixture MakeExporter(EventScript script) std::int64_t FlushUs(logs_exporter::ElasticsearchLogRecordExporter &exporter) { const auto started = std::chrono::steady_clock::now(); - exporter.ForceFlush(std::chrono::milliseconds{20}); + exporter.ForceFlush(kFlushTimeout); return std::chrono::duration_cast(std::chrono::steady_clock::now() - started) .count(); From 8ffbea62d7aaf305582e32186aed8c94092b7f75 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Tue, 22 Sep 2026 05:15:24 +0000 Subject: [PATCH 7/7] [CHANGELOG] Describe the two changes the entry left out Retiring the session before the completion callback logs changes what a caller sees: a log handler that calls ForceFlush() from that line used to wait for the export reporting to it, measured at 2000276us. That is a behaviour change and belongs in the entry rather than only in the pull request body. The completion line also says log record(s) rather than trace span(s) now, which is observable output. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f7329240b..c5b3630762 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,13 @@ Increment the: * [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.29.0] 2026-09-13