From 68a9d861f4b093b9bf5ffc03e6fbf658b55a138f Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Tue, 11 Aug 2026 04:07:37 +0000 Subject: [PATCH 1/9] [BUG] Stop the Elasticsearch async ForceFlush reporting success without waiting ForceFlush() returned true straight away when async export was enabled, so a caller had no way to know whether anything had been delivered. It now waits on the sessions that were in flight when it was called. Sessions are identified rather than counted: each export takes the next id and joins a set, and the wait ends when no id below the entry watermark is left, so a completion cannot satisfy a flush that started after it. Counting alone lets a later export stand in for an earlier one. The wait uses one deadline for the call, so a wakeup that is not a completion resumes against what is left instead of restarting the wait, and the result is the predicate rather than the leftover duration. The ids are uint64_t rather than size_t. The wait compares them by order, which only holds while they keep increasing, and a 32 bit counter reaches its end in days at a rate this exporter is meant to sustain; past that the next watermark is small enough for a still running session to satisfy it. AsyncResponseHandler reports at most once, through a compare and exchange. 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. An export registers its session before anything that can return, so a flush asked from the moment the records arrive waits for them, and a guard reports through the same completion on every early exit. Without it a return added later would strand a waiter on a session that can never finish. What a true return means is documented on the declaration, and it is weaker than "the session ended". Both completion paths publish the outcome before the session is torn down: OnResponse calls CompleteOnce() ahead of its logging, and the handler destructor calls it ahead of FinishSession(). So a true return means every snapshotted export has reported an outcome, and transport cleanup may still be running. Reported in #4336 Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 3 + exporters/elasticsearch/CMakeLists.txt | 6 + .../elasticsearch/es_log_record_exporter.h | 25 +- .../src/es_log_record_exporter.cc | 272 ++++-- .../test/es_log_record_exporter_test.cc | 871 ++++++++++++++++++ 5 files changed, 1083 insertions(+), 94 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a4b17ef3fc..63ad48b8ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -81,6 +81,9 @@ Increment the: deprecated C headers (`stdint.h`, `stddef.h`, `stdlib.h`, `string.h`, `stdio.h`, `ctype.h`, `limits.h`, `assert.h`) with their C++ equivalents ([#4349](https://github.com/open-telemetry/opentelemetry-cpp/pull/4349)) +* [BUG] Stop the Elasticsearch async ForceFlush reporting success without + waiting for the sessions it was asked about + [#4337](https://github.com/open-telemetry/opentelemetry-cpp/pull/4337) * [CONFIGURATION] Add SDK component builder interfaces to the registry [#4358](https://github.com/open-telemetry/opentelemetry-cpp/issues/4358) diff --git a/exporters/elasticsearch/CMakeLists.txt b/exporters/elasticsearch/CMakeLists.txt index 18b2bcd898..f81248fd26 100644 --- a/exporters/elasticsearch/CMakeLists.txt +++ b/exporters/elasticsearch/CMakeLists.txt @@ -55,4 +55,10 @@ if(OTELCPP_BUILD_TESTING) TARGET es_log_record_exporter_test TEST_PREFIX exporter. TEST_LIST es_log_record_exporter_test) + + # AnIndefiniteFlushReturnsOnceEverythingIsFinished exercises the branch that + # waits without a deadline, so a regression there does not fail, it stops. + # CTest's default bound is 25 minutes, which is a long time to spend learning + # that. + set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 30) endif() # OTELCPP_BUILD_TESTING diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h index 6234df59a2..820194ca39 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h @@ -16,8 +16,9 @@ #ifdef ENABLE_ASYNC_EXPORT # include -# include +# include # include +# include #endif OPENTELEMETRY_BEGIN_NAMESPACE @@ -122,8 +123,16 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo /** * Force flush the exporter. + * + * Waits for the asynchronous sessions already started when the call was made, bounded by the + * caller's timeout rather than by the exporter's response timeout. + * * @param timeout an option timeout, default to max. - * @return return true when all data are exported, and false when timeout + * @return true when each of those exports has reported a terminal outcome, false on timeout. + * The outcome is published before the session is torn down, so a true return does not + * mean FinishSession() has run. Nor does it mean the batch reached Elasticsearch: a + * failed export reports too, through the internal log. Surfacing that here is + * [#3075](https://github.com/open-telemetry/opentelemetry-cpp/issues/3075). */ bool ForceFlush( std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override; @@ -149,11 +158,17 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo #ifdef ENABLE_ASYNC_EXPORT struct SynchronizationData { - std::atomic session_counter_{0}; - std::atomic finished_session_counter_{0}; + // Identified rather than counted, so a completion can only satisfy its own waiter. Guarded by + // force_flush_cv_m, the mutex the wait uses, so starting and finishing cannot interleave with a + // waiter's snapshot or predicate. + // + // Sized independently of the platform's size_t. The wait compares ids by order, which only + // holds while they keep increasing, and a 32 bit counter reaches its end in days at a rate + // this exporter is meant to sustain. + std::uint64_t next_session_id{0}; + std::set running_sessions; std::condition_variable force_flush_cv; std::mutex force_flush_cv_m; - std::recursive_mutex force_flush_m; }; nostd::shared_ptr synchronization_data_; #endif diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index af819c8eb7..7939120a7a 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -5,7 +5,7 @@ #include #include #include -#include +#include // IWYU pragma: keep #include // IWYU pragma: keep #include #include @@ -27,9 +27,13 @@ #include "opentelemetry/sdk/logs/recordable.h" #include "opentelemetry/version.h" +// Half this file only exists under ENABLE_ASYNC_EXPORT, and include-what-you-use asks for these +// in the configurations that build it and asks for them to go in the ones that do not. #ifdef ENABLE_ASYNC_EXPORT # include # include +# include + # include "opentelemetry/common/timestamp.h" #endif @@ -267,7 +271,31 @@ class AsyncResponseHandler : public http_client::EventHandler /** * Cleans up the session in the destructor. */ - ~AsyncResponseHandler() override { session_->FinishSession(); } + ~AsyncResponseHandler() override + { + // A handler that goes away without an outcome would leave ForceFlush() waiting on a session + // that can no longer finish. Report before tearing the session down, since FinishSession() + // 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 +303,95 @@ 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; + // Every state is listed so that -Wswitch reports a new one rather than it being swallowed by + // a default label and silently leaving the session uncounted. + const char *failure = nullptr; switch (state) { 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::Created: + OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session created"); + break; + case http_client::SessionState::Destroyed: + failure = "[ES Log Exporter] Session to elasticsearch destroyed before a response"; + break; + case http_client::SessionState::Connecting: + OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connecting to elasticsearch"); 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::Connected: + OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connected to elasticsearch"); + break; + case http_client::SessionState::Sending: + OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Sending request to elasticsearch"); 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::Response: + // The body arrives through OnResponse(), which is what reports the outcome. + OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Response received from 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; - default: + case http_client::SessionState::WriteError: + failure = "[ES Log Exporter] Write error"; + break; + case http_client::SessionState::Cancelled: + failure = "[ES Log Exporter] Request to elasticsearch cancelled"; break; } - if (need_stop) + + // Reported only when this event is the outcome. Any of these can arrive after a response has + // already been reported, and an error line there would describe a failure the exporter never + // told the caller about. + if (failure != nullptr && CompleteOnce(sdk::common::ExportResult::kFailure)) { - result_callback_(sdk::common::ExportResult::kFailure); + OTEL_INTERNAL_LOG_ERROR(failure); } } @@ -344,8 +401,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; @@ -378,12 +435,7 @@ ElasticsearchLogRecordExporter::ElasticsearchLogRecordExporter( , synchronization_data_(new SynchronizationData()) #endif -{ -#ifdef ENABLE_ASYNC_EXPORT - synchronization_data_->finished_session_counter_.store(0); - synchronization_data_->session_counter_.store(0); -#endif -} +{} std::unique_ptr ElasticsearchLogRecordExporter::MakeRecordable() noexcept { @@ -393,6 +445,70 @@ std::unique_ptr ElasticsearchLogRecordExporter::MakeRecorda sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( const nostd::span> &records) noexcept { +#ifdef ENABLE_ASYNC_EXPORT + // Registered before anything that can return, so a flush asked from the moment these records + // arrive waits for them, and every exit below reports through the guard. + const std::size_t span_count = records.size(); + auto synchronization_data = synchronization_data_; + + std::uint64_t session_id = 0; + { + std::lock_guard lock(synchronization_data_->force_flush_cv_m); + session_id = synchronization_data_->next_session_id++; + synchronization_data_->running_sessions.insert(session_id); + } + + using Completion = std::function; + Completion complete = [span_count, session_id, + synchronization_data](opentelemetry::sdk::common::ExportResult result) { + { + // Published under the mutex ForceFlush() waits on. A waiter that has evaluated its + // predicate but not yet parked would otherwise not see this until the next wakeup. + std::lock_guard lock(synchronization_data->force_flush_cv_m); + synchronization_data->running_sessions.erase(session_id); + } + synchronization_data->force_flush_cv.notify_all(); + + // Logged after the session is retired. The log handler is replaceable, and one that calls + // ForceFlush() would otherwise wait for the very session this call has not let go of yet. + if (result != opentelemetry::sdk::common::ExportResult::kSuccess) + { + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export " + << span_count + << " log record(s) error: " << static_cast(result)); + } + else + { + OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count + << " log record(s) success"); + } + return true; + }; + + // A return added between here and SendRequest() would otherwise strand a waiter on a session + // that can never finish. Reporting through the same completion leaves a session one way out, + // and says so in the log rather than dropping the batch silently. + struct GiveUpGuard + { + Completion *report = nullptr; + + GiveUpGuard() = default; + GiveUpGuard(const GiveUpGuard &) = delete; + GiveUpGuard &operator=(const GiveUpGuard &) = delete; + GiveUpGuard(GiveUpGuard &&) = delete; + GiveUpGuard &operator=(GiveUpGuard &&) = delete; + + ~GiveUpGuard() + { + if (report != nullptr) + { + (*report)(opentelemetry::sdk::common::ExportResult::kFailure); + } + } + } guard; + guard.report = &complete; +#endif + // Return failure if this exporter has been shutdown if (isShutdown()) { @@ -436,30 +552,10 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( #ifdef ENABLE_ASYNC_EXPORT // Send the request - synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release); - std::size_t span_count = records.size(); - auto synchronization_data = synchronization_data_; - auto handler = std::make_shared( - session, - [span_count, synchronization_data](opentelemetry::sdk::common::ExportResult result) { - 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)); - } - else - { - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count - << " trace span(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_); + auto handler = std::make_shared(session, Completion(complete), + options_.console_debug_); session->SendRequest(handler); + guard.report = nullptr; // the handler reports this session from here on return sdk::common::ExportResult::kSuccess; #else // Send the request @@ -503,41 +599,39 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou OPENTELEMETRY_MAYBE_UNUSED) noexcept { #ifdef ENABLE_ASYNC_EXPORT - std::lock_guard lock_guard{synchronization_data_->force_flush_m}; - std::size_t running_counter = - synchronization_data_->session_counter_.load(std::memory_order_acquire); // ASAN will report chrono: runtime error: signed integer overflow: A + B cannot be represented - // in type 'long int' here. So we reset timeout to meet signed long int limit here. + // in type 'long int' here. So we reset timeout to meet signed long int limit here. Zero is + // what that returns for a timeout there is no point waiting against, which is also how a + // caller asks for no deadline. timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout( timeout, std::chrono::microseconds::zero()); - std::chrono::steady_clock::duration timeout_steady = - std::chrono::duration_cast(timeout); - if (timeout_steady <= std::chrono::steady_clock::duration::zero()) - { - timeout_steady = (std::chrono::steady_clock::duration::max)(); - } + std::unique_lock lock(synchronization_data_->force_flush_cv_m); - std::unique_lock lk_cv(synchronization_data_->force_flush_cv_m); - // Wait for all the sessions to finish - while (timeout_steady > std::chrono::steady_clock::duration::zero()) - { - if (synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >= - running_counter) - { - break; - } + // The snapshot is the next id, not a count: a session started after it takes a larger id and + // cannot stand in for one of these. Ids are issued in order, so the smallest one still running + // decides. Callers are not serialised, so two deadlines never queue behind one another. + const std::uint64_t watermark = synchronization_data_->next_session_id; + const auto flushed = [this, watermark]() { + const auto &running = synchronization_data_->running_sessions; + return running.empty() || *running.begin() >= watermark; + }; - std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now(); - if (std::cv_status::no_timeout != synchronization_data_->force_flush_cv.wait_for( - lk_cv, std::chrono::seconds{options_.response_timeout_})) - { - break; - } - timeout_steady -= std::chrono::steady_clock::now() - start_timepoint; + if (timeout <= std::chrono::microseconds::zero()) + { + // wait() only returns once the predicate holds, so the flush has completed. + synchronization_data_->force_flush_cv.wait(lock, flushed); + return true; } - return timeout_steady > std::chrono::steady_clock::duration::zero(); + // One deadline for the call, so a wakeup that is not a completion resumes against what is left + // rather than starting the wait again. wait_until() returns the predicate, so a flush that ran + // out of time cannot report success. + const std::chrono::steady_clock::time_point deadline = + std::chrono::steady_clock::now() + + std::chrono::duration_cast(timeout); + + return synchronization_data_->force_flush_cv.wait_until(lock, deadline, flushed); #else return true; #endif diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index a65c0b4c1c..25a0772b79 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -4,11 +4,14 @@ #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h" #include "opentelemetry/common/timestamp.h" #include "opentelemetry/exporters/elasticsearch/es_log_recordable.h" +#include "opentelemetry/ext/http/client/http_client.h" #include "opentelemetry/logs/severity.h" +#include "opentelemetry/nostd/function_ref.h" #include "opentelemetry/nostd/span.h" #include "opentelemetry/nostd/string_view.h" #include "opentelemetry/nostd/utility.h" #include "opentelemetry/sdk/common/exporter_utils.h" +#include "opentelemetry/sdk/common/global_log_handler.h" #include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h" #include "opentelemetry/sdk/logs/exporter.h" #include "opentelemetry/sdk/logs/recordable.h" @@ -16,17 +19,25 @@ #include #include +#include #include +#include #include #include +#include +#include +#include #include +#include #include +#include #include "nlohmann/json.hpp" namespace sdklogs = opentelemetry::sdk::logs; namespace logs_api = opentelemetry::logs; namespace nostd = opentelemetry::nostd; namespace logs_exporter = opentelemetry::exporter::logs; +namespace internal_log = opentelemetry::sdk::common::internal_log; TEST(ElasticsearchLogsExporterTests, CustomClientConstructionSucceeds) { @@ -142,3 +153,863 @@ 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; + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + EventScript script_; +}; + +} // namespace + +// --------------------------------------------------------------------------- +// ForceFlush deadline. Only built with async export, which is the only +// configuration where the wait exists. +// --------------------------------------------------------------------------- +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 + +// The wait these cases describe exists only in an async build, so they skip elsewhere rather than +// compile out: gtest_add_tests reads the source, and a case missing from the binary is still +// registered with CTest, where it then reports a pass without having run. The skip goes in SetUp +// because GTEST_SKIP returns, and a skip at the top of each body would leave the rest of that body +// unreachable, which MSVC reports as C4702. +namespace +{ +class ElasticsearchForceFlushTests : public ::testing::Test +{ +protected: + void SetUp() override + { +#ifndef ENABLE_ASYNC_EXPORT + GTEST_SKIP() << "ForceFlush has nothing to wait for when async export is disabled"; +#endif + } +}; +} // namespace + +// The session never calls back, so the flush cannot complete. It has to say so, and it has to say +// so when the caller's deadline runs out rather than when the response timeout does. +TEST_F(ElasticsearchForceFlushTests, ReportsFailureWhenTheFlushDoesNotComplete) +{ + // The handler is kept because a dropped one now reports a failure from its destructor, which + // would finish the session and leave nothing for the flush to wait on. The real curl operation + // owns the handler until the request ends, so this is also the truer shape. + std::vector> kept; + auto fixture = MakeExporter([&kept](const std::shared_ptr &handler) { + kept.push_back(handler); + }); + ExportOnce(*fixture.exporter); + + const auto start = std::chrono::steady_clock::now(); + const bool flushed = fixture.exporter->ForceFlush(std::chrono::milliseconds{20}); + const auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_FALSE(flushed); + EXPECT_LT(std::chrono::duration_cast(elapsed).count(), + kShortResponseTimeoutSeconds * 1000); +} + +// Nothing outstanding, so there is nothing to wait for. +TEST_F(ElasticsearchForceFlushTests, ReturnsImmediatelyWithNothingInFlight) +{ + auto fixture = MakeExporter([](const std::shared_ptr &) {}); + + const auto start = std::chrono::steady_clock::now(); + EXPECT_TRUE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); + EXPECT_LT(std::chrono::duration_cast(std::chrono::steady_clock::now() - + start) + .count(), + kShortResponseTimeoutSeconds * 1000); +} + +// The session completes inside SendRequest(), before the flush is even asked for, so the +// completion is already published when the predicate is first evaluated. +TEST_F(ElasticsearchForceFlushTests, SucceedsWhenTheSessionFinishedBeforeTheWait) +{ + auto fixture = MakeExporter([](const std::shared_ptr &handler) { + FakeResponse response(200, kAcceptedBody); + handler->OnResponse(response); + }); + ExportOnce(*fixture.exporter); + + EXPECT_TRUE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); +} + +// Two exports, one of which never finishes. Reporting success here would tell the caller data was +// flushed that is still in flight. +TEST_F(ElasticsearchForceFlushTests, PartialCompletionIsNotSuccess) +{ + bool respond = true; + std::vector> kept; + auto fixture = + MakeExporter([&respond, &kept](const std::shared_ptr &handler) { + kept.push_back(handler); // the second session has to stay unfinished + if (respond) + { + FakeResponse response(200, kAcceptedBody); + handler->OnResponse(response); + } + }); + ExportOnce(*fixture.exporter); + respond = false; + ExportOnce(*fixture.exporter); + + EXPECT_FALSE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); +} + +// A completion delivered from another thread after the waiter has parked has to wake it. This is +// the half the inline scripts above cannot reach. +TEST_F(ElasticsearchForceFlushTests, ACompletionAfterTheWaiterParksWakesIt) +{ + // Held by the test, not by the fakes: a handler owns its session, so a session that also owned + // its handler would be a reference cycle and leak. + std::shared_ptr captured; + auto fixture = + MakeExporter([&captured](const std::shared_ptr &handler) { + captured = handler; + }); + ExportOnce(*fixture.exporter); + ASSERT_NE(captured, nullptr); + + std::thread responder([&captured] { + std::this_thread::sleep_for(std::chrono::milliseconds{50}); + FakeResponse response(200, kAcceptedBody); + captured->OnResponse(response); + }); + + const bool flushed = fixture.exporter->ForceFlush(std::chrono::seconds{5}); + responder.join(); + EXPECT_TRUE(flushed); +} + +// A session started after the flush was asked for belongs to a later batch, so its completion is +// not evidence about the batch the caller is waiting on. Counting completions rather than +// identifying them let one stand in for the other: the flush reported success with the original +// export still in flight. +// +// The ordering the case needs is that the flush takes its snapshot before the second export +// starts. ForceFlush() is called with nothing between it and the thread that starts, and the other +// side sleeps first, so the window is four orders of magnitude wider than the race. Reverting the +// production change makes this fail, which is what shows the ordering held. +TEST_F(ElasticsearchForceFlushTests, ANewerSessionDoesNotStandInForAnOlderOne) +{ + std::vector> kept; + std::shared_ptr latest; + auto fixture = + MakeExporter([&kept, &latest](const std::shared_ptr &handler) { + kept.push_back(handler); + latest = handler; + }); + + // The batch the caller waits for. It never finishes. + ExportOnce(*fixture.exporter); + + std::thread newer([&fixture, &latest] { + std::this_thread::sleep_for(std::chrono::milliseconds{100}); + ExportOnce(*fixture.exporter); + FakeResponse response(200, kAcceptedBody); + latest->OnResponse(response); + }); + + const bool flushed = fixture.exporter->ForceFlush(std::chrono::milliseconds{500}); + newer.join(); + + EXPECT_FALSE(flushed) << "a later batch's completion flushed an export that is still in flight"; +} + +// A batch counts as received when Export() is entered, not when its request has finished being +// built, so a flush asked while the body is still being serialised has to wait for it. Registering +// the session after the request was built left a window where the flush snapshotted past it and +// reported it as already flushed. CreateSession() stands in for the serialisation: it runs after +// the records were handed over and before the request exists. +TEST_F(ElasticsearchForceFlushTests, AnExportAlreadyUnderWayIsSomethingToWaitFor) +{ + std::vector> kept; + auto fixture = MakeExporter([&kept](const std::shared_ptr &handler) { + kept.push_back(handler); + }); + + std::mutex mutex; + std::condition_variable cv; + bool inside_export = false; + bool release = false; + fixture.client->on_create_session = [&mutex, &cv, &inside_export, &release] { + { + std::lock_guard lock(mutex); + inside_export = true; + } + cv.notify_all(); + std::unique_lock lock(mutex); + cv.wait(lock, [&release] { return release; }); + }; + + std::thread exporting([&fixture] { ExportOnce(*fixture.exporter); }); + { + std::unique_lock lock(mutex); + cv.wait(lock, [&inside_export] { return inside_export; }); + } + + const bool flushed = fixture.exporter->ForceFlush(std::chrono::milliseconds{200}); + + { + std::lock_guard lock(mutex); + release = true; + } + cv.notify_all(); + exporting.join(); + + EXPECT_FALSE(flushed) << "the flush passed over a batch Export() had already been handed"; +} + +// The same substitution with the count actually reaching the snapshot, which is what makes a +// counter look right while being wrong. Two exports are outstanding when the flush is asked, so it +// waits for two completions; a third export starts after that, and it plus one of the originals +// deliver two completions. The export the caller is waiting on has still not finished. +TEST_F(ElasticsearchForceFlushTests, LaterCompletionsCannotCoverAnOlderOutstandingExport) +{ + std::vector> kept; + std::shared_ptr latest; + auto fixture = + MakeExporter([&kept, &latest](const std::shared_ptr &handler) { + kept.push_back(handler); + latest = handler; + }); + + ExportOnce(*fixture.exporter); // never finishes + ExportOnce(*fixture.exporter); + auto second = latest; + + std::thread worker([&fixture, &second, &latest] { + std::this_thread::sleep_for(std::chrono::milliseconds{100}); + ExportOnce(*fixture.exporter); // starts after the flush took its snapshot + const auto &third = latest; + FakeResponse response(200, kAcceptedBody); + second->OnResponse(response); + third->OnResponse(response); + }); + + const bool flushed = fixture.exporter->ForceFlush(std::chrono::milliseconds{500}); + worker.join(); + + EXPECT_FALSE(flushed) << "two completions arrived, but not the one the caller was waiting for"; +} + +// A second caller gets its own deadline. Serialising the calls is fine, making the second one +// wait out the first one's is not, since its timeout would mean nothing. +TEST_F(ElasticsearchForceFlushTests, AConcurrentFlushKeepsItsOwnDeadline) +{ + std::vector> kept; + auto fixture = MakeExporter([&kept](const std::shared_ptr &handler) { + kept.push_back(handler); + }); + ExportOnce(*fixture.exporter); + + // The first caller waits well past the bound asserted below, so a second caller that queued + // behind it could not come in under that bound by accident. + std::thread slow([&fixture] { fixture.exporter->ForceFlush(std::chrono::milliseconds{1500}); }); + std::this_thread::sleep_for(std::chrono::milliseconds{100}); + + const auto start = std::chrono::steady_clock::now(); + EXPECT_FALSE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); + const auto ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); + slow.join(); + + EXPECT_LT(ms, 700) << "waited behind the first caller instead of its own deadline"; +} + +// The default argument is microseconds::max(), which AdjustWaitForTimeout maps to the sentinel for +// no deadline. That branch takes the lock outright and waits on the predicate, so it needs a case +// where the predicate already holds or the test would never return. +TEST_F(ElasticsearchForceFlushTests, AnIndefiniteFlushReturnsOnceEverythingIsFinished) +{ + auto fixture = MakeExporter([](const std::shared_ptr &handler) { + FakeResponse response(200, kAcceptedBody); + handler->OnResponse(response); + }); + ExportOnce(*fixture.exporter); + + EXPECT_TRUE(fixture.exporter->ForceFlush()); +} + +// What true means, pinned so that it cannot drift unnoticed: the snapshotted exports have +// reported an outcome, not that their batches were delivered. Elasticsearch rejected this one +// and the flush still reports success. Surfacing an export failure through this return value is +// https://github.com/open-telemetry/opentelemetry-cpp/issues/3075, which is a repository wide +// decision rather than one for this exporter: the OTLP HTTP client's ForceFlush reports the same +// way today, and changing one of them alone would leave the two disagreeing. +TEST_F(ElasticsearchForceFlushTests, AFailedExportSettlesItsSessionAndIsReportedAsFlushed) +{ + auto fixture = MakeExporter([](const std::shared_ptr &handler) { + FakeResponse response(200, R"({"took":1,"errors":true,"items":[]})"); + handler->OnResponse(response); + }); + ExportOnce(*fixture.exporter); + + EXPECT_TRUE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); +} +// --------------------------------------------------------------------------- +// 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; + } + 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(); } + +private: + std::atomic successes_{0}; + std::atomic failures_{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. + 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( + nostd::shared_ptr(new internal_log::DefaultLogHandler())); + } + } + + const CompletionCountingLogHandler &Counter() const + { + return *static_cast(handler_.get()); + } + + int Completions() const { return Counter().completions(); } + + nostd::shared_ptr handler_; + internal_log::LogLevel previous_level_ = internal_log::LogLevel::Warning; +}; +} // namespace +// Every terminal state has to finish the session, or a flush waits on a request that can never +// complete. Every progress state has to leave it running. +TEST_F(ElasticsearchAsyncCompletionTests, EverySessionStateIsClassifiedAndReportsAtMostOnce) +{ + struct Case + { + http_client::SessionState state; + bool terminal; + }; + + // Every member of the enum, so a state added upstream shows up here as a missing row rather than + // as a session that quietly never finishes. The switch has no default label for the same reason. + const Case cases[] = { + {http_client::SessionState::CreateFailed, true}, + {http_client::SessionState::Created, false}, + {http_client::SessionState::Destroyed, true}, + {http_client::SessionState::Connecting, false}, + {http_client::SessionState::ConnectFailed, true}, + {http_client::SessionState::Connected, false}, + {http_client::SessionState::Sending, false}, + {http_client::SessionState::SendFailed, true}, + {http_client::SessionState::Response, false}, + {http_client::SessionState::SSLHandshakeFailed, true}, + {http_client::SessionState::TimedOut, true}, + {http_client::SessionState::NetworkError, true}, + {http_client::SessionState::ReadError, true}, + {http_client::SessionState::WriteError, true}, + {http_client::SessionState::Cancelled, true}, + }; + + for (const auto &test_case : cases) + { + SCOPED_TRACE(static_cast(test_case.state)); + std::vector> kept; + auto fixture = MakeExporter( + [&kept, &test_case](const std::shared_ptr &handler) { + kept.push_back(handler); // so the event, not the destructor, is what decides + handler->OnEvent(test_case.state, ""); + handler->OnEvent(test_case.state, ""); // a repeat must not report a second time + }); + + // Counted per iteration: the previous fixture's handler reports from its destructor as it goes + // out of scope, which lands in the same counter. + const int before = Completions(); + ExportOnce(*fixture.exporter); + + EXPECT_EQ(Completions() - before, test_case.terminal ? 1 : 0); + EXPECT_EQ(fixture.exporter->ForceFlush(std::chrono::milliseconds{20}), test_case.terminal); + } +} + +// 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); + } +} + +// A response decides the outcome, and a teardown event arriving after it must not report again. +// Both orderings, because the first verdict is the one that has to survive either way. +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"; + } +} + +// 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: + void Watch(logs_exporter::ElasticsearchLogRecordExporter *exporter) noexcept + { + exporter_ = exporter; + } + + 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("Logs were not written") == std::string::npos) + { + return; + } + 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); } + +private: + logs_exporter::ElasticsearchLogRecordExporter *exporter_{nullptr}; + std::atomic reentered_{false}; + std::atomic flushed_{false}; +}; +} // 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); +} + +// 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"; +} + +TEST_F(ElasticsearchAsyncCompletionTests, TwoTerminalEventsCountAsOneSession) +{ + // The real curl operation owns the handler until the request finishes, so a fake that lets it + // go would make every request complete the moment SendRequest returns. The test owns them + // instead of the fakes, since a handler owns its session and the reverse would be a cycle. + std::vector> kept; + int session = 0; + auto fixture = + MakeExporter([&kept, &session](const std::shared_ptr &handler) { + kept.push_back(handler); + if (++session == 1) + { + handler->OnEvent(http_client::SessionState::ConnectFailed, ""); + handler->OnEvent(http_client::SessionState::CreateFailed, ""); + } + }); + + ExportOnce(*fixture.exporter); // reports twice before the fix + ExportOnce(*fixture.exporter); // never calls back + + EXPECT_EQ(Completions(), 1) << "the first session reported more than once"; + EXPECT_FALSE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})) + << "the second session is still in flight"; +} + +// The other in-tree double fire: the completion lambda uses two independent ifs, so an aborted +// operation that also has a response reports Cancelled and then delivers the response. +TEST_F(ElasticsearchAsyncCompletionTests, CancelledThenAResponseCountsOnce) +{ + std::vector> kept; + int session = 0; + auto fixture = + MakeExporter([&kept, &session](const std::shared_ptr &handler) { + kept.push_back(handler); + if (++session == 1) + { + handler->OnEvent(http_client::SessionState::Cancelled, ""); + FakeResponse response(200, kAcceptedBody); + handler->OnResponse(response); + } + }); + + ExportOnce(*fixture.exporter); + ExportOnce(*fixture.exporter); + + EXPECT_EQ(Completions(), 1) << "the late response reported on top of the cancellation"; + EXPECT_EQ(Counter().successes(), 0) << "the late response replaced the cancellation verdict"; + EXPECT_FALSE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); +} + +// A response followed by the session being torn down is the ordinary successful shape. +TEST_F(ElasticsearchAsyncCompletionTests, ResponseThenDestroyedCountsOnce) +{ + std::vector> kept; + int session = 0; + auto fixture = + MakeExporter([&kept, &session](const std::shared_ptr &handler) { + kept.push_back(handler); + if (++session == 1) + { + FakeResponse response(200, kAcceptedBody); + handler->OnResponse(response); + handler->OnEvent(http_client::SessionState::Destroyed, ""); + } + }); + + ExportOnce(*fixture.exporter); + ExportOnce(*fixture.exporter); + + EXPECT_EQ(Completions(), 1) << "the teardown reported on top of the response"; + EXPECT_FALSE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); +} + +// 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 95c1a09da774378b272bed4b62cafea258722492 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Thu, 13 Aug 2026 18:25:30 +0000 Subject: [PATCH 2/9] [BUG] Retire a refused export before describing the refusal The export is registered before the shutdown check, and the guard that retires it could only run after the refusal had already been described. The log handler is replaceable, so a handler that calls ForceFlush() from that error waited for the Export() that was calling it. The same path also described one refusal twice, once itself and once through the guard. Retiring is now separate from reporting. The refusal retires, disarms the guard, and says so once. The completion still answers true, because changing what it answers would change what the response handler reads. Measured with a handler that calls ForceFlush(20ms) from the shutdown error. Before: the flush returns false after waiting out its full 20 ms, 3 of 3. After: it returns true in 0 ms, 3 of 3, and one refusal reads as one line. The file holds at 25 passing, 3 of 3. Three statements alongside it claimed more than they hold. An array cannot notice that an enum grew, so the comment now names the switch without a default under -Wswitch as what catches a new state. The curl completion lambda tests the response first and the abort in an else, so the comment that called it two independent ifs is out of date. And the fixture put a fresh default log handler back rather than the one it found. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../src/es_log_record_exporter.cc | 22 +++++++-- .../test/es_log_record_exporter_test.cc | 48 +++++++++++++++---- 2 files changed, 58 insertions(+), 12 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 7939120a7a..6aa3f2c523 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -458,16 +458,24 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( synchronization_data_->running_sessions.insert(session_id); } - using Completion = std::function; - Completion complete = [span_count, session_id, - synchronization_data](opentelemetry::sdk::common::ExportResult result) { + // Retiring the export is separate from describing what happened to it, because the refusal + // below has to retire without having an outcome to report through the completion. + auto retire = [session_id, synchronization_data]() noexcept { + bool retired = false; { // Published under the mutex ForceFlush() waits on. A waiter that has evaluated its // predicate but not yet parked would otherwise not see this until the next wakeup. std::lock_guard lock(synchronization_data->force_flush_cv_m); - synchronization_data->running_sessions.erase(session_id); + retired = synchronization_data->running_sessions.erase(session_id) == 1; } synchronization_data->force_flush_cv.notify_all(); + return retired; + }; + + using Completion = std::function; + Completion complete = [span_count, + retire](opentelemetry::sdk::common::ExportResult result) noexcept { + retire(); // Logged after the session is retired. The log handler is replaceable, and one that calls // ForceFlush() would otherwise wait for the very session this call has not let go of yet. @@ -512,6 +520,12 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( // Return failure if this exporter has been shutdown if (isShutdown()) { +#ifdef ENABLE_ASYNC_EXPORT + // Retired before anything replaceable runs, and reported here rather than through the guard, + // so a log handler that flushes does not wait for this Export() and one refusal reads as one. + guard.report = nullptr; + retire(); +#endif OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting " << records.size() << " log(s) failed, exporter is shutdown"); return sdk::common::ExportResult::kFailure; diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 25a0772b79..72e4f5b213 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -626,6 +626,7 @@ class ElasticsearchAsyncCompletionTests : public ::testing::Test #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(); @@ -638,8 +639,7 @@ class ElasticsearchAsyncCompletionTests : public ::testing::Test if (handler_) { internal_log::GlobalLogHandler::SetLogLevel(previous_level_); - internal_log::GlobalLogHandler::SetLogHandler( - nostd::shared_ptr(new internal_log::DefaultLogHandler())); + internal_log::GlobalLogHandler::SetLogHandler(previous_handler_); } } @@ -651,6 +651,7 @@ class ElasticsearchAsyncCompletionTests : public ::testing::Test int Completions() const { return Counter().completions(); } nostd::shared_ptr handler_; + nostd::shared_ptr previous_handler_; internal_log::LogLevel previous_level_ = internal_log::LogLevel::Warning; }; } // namespace @@ -664,8 +665,9 @@ TEST_F(ElasticsearchAsyncCompletionTests, EverySessionStateIsClassifiedAndReport bool terminal; }; - // Every member of the enum, so a state added upstream shows up here as a missing row rather than - // as a session that quietly never finishes. The switch has no default label for the same reason. + // Every member of the enum as it stands. An array cannot notice that the enum grew, so what + // catches a new state is the switch in the exporter having no default, under the -Wswitch that + // maintainer mode turns into an error. This table is the second half of that. const Case cases[] = { {http_client::SessionState::CreateFailed, true}, {http_client::SessionState::Created, false}, @@ -846,9 +848,13 @@ namespace class FlushingLogHandler : public internal_log::LogHandler { public: - void Watch(logs_exporter::ElasticsearchLogRecordExporter *exporter) noexcept + // 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 */, @@ -861,10 +867,11 @@ class FlushingLogHandler : public internal_log::LogHandler { return; } - if (std::string(msg).find("Logs were not written") == std::string::npos) + 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; @@ -874,11 +881,14 @@ class FlushingLogHandler : public internal_log::LogHandler 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 @@ -903,6 +913,27 @@ TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromInsideTheLogHandlerDoesNotWa 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) @@ -958,8 +989,9 @@ TEST_F(ElasticsearchAsyncCompletionTests, TwoTerminalEventsCountAsOneSession) << "the second session is still in flight"; } -// The other in-tree double fire: the completion lambda uses two independent ifs, so an aborted -// operation that also has a response reports Cancelled and then delivers the response. +// The curl client's completion lambda tests the response first and the abort in an else, so this +// ordering no longer comes from it. It still comes from anywhere: EventHandler promises callers +// nothing about how many terminal events arrive, and #4360 is open on exactly that. TEST_F(ElasticsearchAsyncCompletionTests, CancelledThenAResponseCountsOnce) { std::vector> kept; From e73725b1644b6482850e8aaf6b0d09899ff6f24e Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Fri, 14 Aug 2026 22:33:21 +0000 Subject: [PATCH 3/9] [TEST] Make the snapshot ordering a fact, and hold the two boundaries it hid Three of the cases here needed an export to start after a flush had taken its watermark, and said so with a sleep. That leaves the order to the scheduler, and the order a delayed runner picks is the one where the newer export is inside the snapshot, where even the counting model these cases exist to rule out reports a pass. A sleep that is too short is not flaky here, it is green for the wrong reason. So the flush raises a counter under the mutex it already holds, immediately after reading its watermark, and the cases wait on that. Nothing in the exporter reads it. The wait is bounded and the count only goes up, so a wait that expires means the flush never got there rather than that the case missed it. Two boundaries had no case at all, and mutating the predicate showed it. All three of these left the suite green at 25 of 25: if (timeout <= 0) { return flushed(); } the indefinite wait return running.empty(); the whole watermark return running.empty() || *running.begin() > watermark; The first is the no-deadline branch, which the case named for it never reaches: the fake answers from inside SendRequest(), so the session is gone and the predicate holds before ForceFlush() is called. The other two are the opposite boundary from the substitution cases: those hold that a newer completion cannot finish an older flush, and nothing held that a newer export still running cannot keep that flush open. AnIndefiniteFlushParksUntilTheOutcomeArrives waits on the indefinite branch with nothing answering, checks it has not returned, then answers. ANewerSessionDoesNotHoldAnOlderFlushOpen starts a second export after the watermark, answers only the first, and requires the flush to return. Neither asserts anything fatal while its waiter thread is running. A fatal assertion there returns from the body with the thread still joinable, which ends the process: the first version of these two aborted the whole binary rather than reporting one case. All 27 cases pass, three runs out of three. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../elasticsearch/es_log_record_exporter.h | 9 ++ .../src/es_log_record_exporter.cc | 3 +- .../test/es_log_record_exporter_test.cc | 124 ++++++++++++++++++ 3 files changed, 135 insertions(+), 1 deletion(-) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h index 820194ca39..48d9bc7227 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h @@ -169,8 +169,17 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo std::set running_sessions; std::condition_variable force_flush_cv; std::mutex force_flush_cv_m; + + // Raised once per ForceFlush(), under that mutex and immediately after the call has taken + // its watermark. Nothing in the exporter reads it. A case that has to start an export after + // a flush has snapshotted, and before it gives up, waits on this: the alternative is a sleep, + // which leaves the order to the scheduler, and the order that does not reproduce the defect + // is the one that would report a pass. + std::uint64_t watermarks_taken{0}; }; nostd::shared_ptr synchronization_data_; + + friend class ElasticsearchExporterTestPeer; #endif }; } // namespace logs diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 6aa3f2c523..07c06fa3cf 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -626,7 +626,8 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou // cannot stand in for one of these. Ids are issued in order, so the smallest one still running // decides. Callers are not serialised, so two deadlines never queue behind one another. const std::uint64_t watermark = synchronization_data_->next_session_id; - const auto flushed = [this, watermark]() { + ++synchronization_data_->watermarks_taken; + const auto flushed = [this, watermark]() { const auto &running = synchronization_data_->running_sessions; return running.empty() || *running.begin() >= watermark; }; diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 72e4f5b213..dabb69df22 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -254,6 +254,52 @@ class FakeHttpClient : public http_client::HttpClient } // namespace +// --------------------------------------------------------------------------- +// Snapshot ordering, for the cases that need an export to start after a flush has taken its +// watermark. That order decides what the flush is waiting on, so it has to be a fact rather +// than a sleep: on a loaded runner the other order runs instead, and in that order even the +// counting model these cases exist to rule out would report a pass. +// --------------------------------------------------------------------------- +#ifdef ENABLE_ASYNC_EXPORT +OPENTELEMETRY_BEGIN_NAMESPACE +namespace exporter +{ +namespace logs +{ +class ElasticsearchExporterTestPeer +{ +public: + static std::uint64_t WatermarksTaken(ElasticsearchLogRecordExporter &exporter) + { + std::lock_guard lock_guard{exporter.synchronization_data_->force_flush_cv_m}; + return exporter.synchronization_data_->watermarks_taken; + } +}; +} // namespace logs +} // namespace exporter +OPENTELEMETRY_END_NAMESPACE + +namespace +{ +// Bounded, and the count only goes up, so a wait that expires means the flush never got there +// rather than that the case missed it. +bool WaitForWatermarks(logs_exporter::ElasticsearchLogRecordExporter &exporter, + std::uint64_t wanted) +{ + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds{30}; + while (std::chrono::steady_clock::now() < deadline) + { + if (logs_exporter::ElasticsearchExporterTestPeer::WatermarksTaken(exporter) >= wanted) + { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds{1}); + } + return false; +} +} // namespace +#endif // ENABLE_ASYNC_EXPORT + // --------------------------------------------------------------------------- // ForceFlush deadline. Only built with async export, which is the only // configuration where the wait exists. @@ -542,6 +588,84 @@ TEST_F(ElasticsearchForceFlushTests, AConcurrentFlushKeepsItsOwnDeadline) // The default argument is microseconds::max(), which AdjustWaitForTimeout maps to the sentinel for // no deadline. That branch takes the lock outright and waits on the predicate, so it needs a case // where the predicate already holds or the test would never return. +// The indefinite wait, actually waited on. The case above it reaches the same branch but the +// fake answers from inside SendRequest(), so the session is already gone and the predicate holds +// before ForceFlush() is called. Nothing there would notice the wait being replaced by one +// evaluation of the predicate. +TEST_F(ElasticsearchForceFlushTests, AnIndefiniteFlushParksUntilTheOutcomeArrives) +{ + std::shared_ptr captured; + auto fixture = + MakeExporter([&captured](const std::shared_ptr &handler) { + captured = handler; + }); + ExportOnce(*fixture.exporter); + ASSERT_NE(captured, nullptr); + + std::atomic returned{false}; + bool flushed = false; + std::thread waiter([&fixture, &returned, &flushed] { + flushed = fixture.exporter->ForceFlush(); + returned.store(true, std::memory_order_release); + }); + + // Recorded rather than asserted here. A fatal assertion between starting that thread and + // joining it leaves it joinable, and a joinable thread being destroyed ends the process, which + // takes the rest of the binary with it instead of reporting one case. + const bool snapshotted = WaitForWatermarks(*fixture.exporter, 1); + const bool returned_early = returned.load(std::memory_order_acquire); + + FakeResponse response(200, kAcceptedBody); + captured->OnResponse(response); + waiter.join(); + + EXPECT_TRUE(snapshotted) << "the flush never took a watermark, so it never reached the wait"; + EXPECT_FALSE(returned_early) << "the flush returned with its session still outstanding"; + EXPECT_TRUE(flushed) << "the flush did not report the completion it was waiting for"; +} + +// The other boundary from the substitution cases. Those hold that a newer completion cannot +// finish an older flush; this holds that a newer export still running cannot keep that flush +// open. Without it the predicate could become running.empty(), or compare with > instead of >=, +// and nothing would report it. +TEST_F(ElasticsearchForceFlushTests, ANewerSessionDoesNotHoldAnOlderFlushOpen) +{ + std::vector> kept; + auto fixture = MakeExporter([&kept](const std::shared_ptr &handler) { + kept.push_back(handler); + }); + + ExportOnce(*fixture.exporter); + ASSERT_EQ(kept.size(), static_cast(1)); + + std::atomic returned{false}; + bool flushed = false; + std::thread waiter([&fixture, &returned, &flushed] { + flushed = fixture.exporter->ForceFlush(std::chrono::seconds{5}); + returned.store(true, std::memory_order_release); + }); + + // Recorded rather than asserted, for the same reason as the case above: nothing fatal may + // happen while that thread is still running. + const bool snapshotted = WaitForWatermarks(*fixture.exporter, 1); + + // Started after the snapshot and never answered, so it is outside what this flush waits on. + ExportOnce(*fixture.exporter); + const std::size_t exports_started = kept.size(); + const bool returned_early = returned.load(std::memory_order_acquire); + + FakeResponse response(200, kAcceptedBody); + kept.front()->OnResponse(response); + waiter.join(); + + EXPECT_TRUE(snapshotted) + << "the flush never took a watermark, so the second export is not newer than one"; + EXPECT_EQ(exports_started, static_cast(2)); + EXPECT_FALSE(returned_early) + << "the flush returned before the export it snapshotted had answered"; + EXPECT_TRUE(flushed) << "the flush waited out its deadline on an export that started after it"; +} + TEST_F(ElasticsearchForceFlushTests, AnIndefiniteFlushReturnsOnceEverythingIsFinished) { auto fixture = MakeExporter([](const std::shared_ptr &handler) { From e8685cb96cd19816ca050e6f06660f7275f0f38e Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Fri, 14 Aug 2026 22:36:28 +0000 Subject: [PATCH 4/9] Count the destructor, and hold the order the comment claimed The retained handler is alive when these cases read the completion count, so its destructor has not run. A regression that reports a second result from there lands after the check: making the destructor bypass CompleteOnce left the suite green at 25 of 25. Letting the handler go before a second read makes the callback and the destructor together exactly one, and three cases now catch that mutation. The response-then-teardown case said both orderings and had one. The other order is its own case now, which is also where the contract goes: a read or write error settles the export, and a response after it is ignored rather than replacing the verdict. EventHandler does not say whether either state can be followed by a response, so this is the choice this exporter makes, written where a change to it is visible. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../test/es_log_record_exporter_test.cc | 46 ++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index dabb69df22..9670196ebc 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -863,11 +863,18 @@ TEST_F(ElasticsearchAsyncCompletionTests, TerminalOrderingsReportExactlyOnce) 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. -// Both orderings, because the first verdict is the one that has to survive either way. +// 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 : @@ -891,6 +898,43 @@ TEST_F(ElasticsearchAsyncCompletionTests, AResponseAndATeardownEventReportOnce) 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"; } } From 355fdd8f7d4db59974776dce640c1a3a4c3db893 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Fri, 14 Aug 2026 22:38:41 +0000 Subject: [PATCH 5/9] Wait for the snapshot instead of timing it Four cases needed an export, a completion or a second caller to arrive after a flush had taken its watermark, and said so with a 50 or 100 millisecond sleep. That leaves the order to the scheduler. In the other order the newer work is inside the snapshot, and then the counting model these cases exist to rule out reports a pass too, so a runner slow enough to invert them does not make the case flaky, it makes it green for the wrong reason. They wait on the watermark count now. Each records whether the wait held and checks it after the join, because a fatal assertion with one of those threads still running would end the process rather than report the case. One sleep is left, the millisecond poll inside that wait. All 28 cases pass, three runs out of three. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../test/es_log_record_exporter_test.cc | 34 +++++++++++++------ 1 file changed, 24 insertions(+), 10 deletions(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 9670196ebc..122c5b16c2 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -437,14 +437,17 @@ TEST_F(ElasticsearchForceFlushTests, ACompletionAfterTheWaiterParksWakesIt) ExportOnce(*fixture.exporter); ASSERT_NE(captured, nullptr); - std::thread responder([&captured] { - std::this_thread::sleep_for(std::chrono::milliseconds{50}); + bool parked = false; + std::thread responder([&fixture, &captured, &parked] { + parked = WaitForWatermarks(*fixture.exporter, 1); FakeResponse response(200, kAcceptedBody); captured->OnResponse(response); }); const bool flushed = fixture.exporter->ForceFlush(std::chrono::seconds{5}); responder.join(); + + EXPECT_TRUE(parked) << "the flush never took a watermark, so it was not waiting for this"; EXPECT_TRUE(flushed); } @@ -470,8 +473,9 @@ TEST_F(ElasticsearchForceFlushTests, ANewerSessionDoesNotStandInForAnOlderOne) // The batch the caller waits for. It never finishes. ExportOnce(*fixture.exporter); - std::thread newer([&fixture, &latest] { - std::this_thread::sleep_for(std::chrono::milliseconds{100}); + bool snapshotted = false; + std::thread newer([&fixture, &latest, &snapshotted] { + snapshotted = WaitForWatermarks(*fixture.exporter, 1); ExportOnce(*fixture.exporter); FakeResponse response(200, kAcceptedBody); latest->OnResponse(response); @@ -480,6 +484,8 @@ TEST_F(ElasticsearchForceFlushTests, ANewerSessionDoesNotStandInForAnOlderOne) const bool flushed = fixture.exporter->ForceFlush(std::chrono::milliseconds{500}); newer.join(); + EXPECT_TRUE(snapshotted) + << "the second export did not start after the snapshot, so it is not a later batch"; EXPECT_FALSE(flushed) << "a later batch's completion flushed an export that is still in flight"; } @@ -545,8 +551,9 @@ TEST_F(ElasticsearchForceFlushTests, LaterCompletionsCannotCoverAnOlderOutstandi ExportOnce(*fixture.exporter); auto second = latest; - std::thread worker([&fixture, &second, &latest] { - std::this_thread::sleep_for(std::chrono::milliseconds{100}); + bool snapshotted = false; + std::thread worker([&fixture, &second, &latest, &snapshotted] { + snapshotted = WaitForWatermarks(*fixture.exporter, 1); ExportOnce(*fixture.exporter); // starts after the flush took its snapshot const auto &third = latest; FakeResponse response(200, kAcceptedBody); @@ -557,6 +564,8 @@ TEST_F(ElasticsearchForceFlushTests, LaterCompletionsCannotCoverAnOlderOutstandi const bool flushed = fixture.exporter->ForceFlush(std::chrono::milliseconds{500}); worker.join(); + EXPECT_TRUE(snapshotted) + << "the third export did not start after the snapshot, so it is not a later batch"; EXPECT_FALSE(flushed) << "two completions arrived, but not the one the caller was waiting for"; } @@ -573,15 +582,20 @@ TEST_F(ElasticsearchForceFlushTests, AConcurrentFlushKeepsItsOwnDeadline) // The first caller waits well past the bound asserted below, so a second caller that queued // behind it could not come in under that bound by accident. std::thread slow([&fixture] { fixture.exporter->ForceFlush(std::chrono::milliseconds{1500}); }); - std::this_thread::sleep_for(std::chrono::milliseconds{100}); - const auto start = std::chrono::steady_clock::now(); - EXPECT_FALSE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); - const auto ms = std::chrono::duration_cast( + // Recorded rather than asserted: that thread is still running, and a fatal assertion here + // would destroy it while it is joinable, which ends the process. + const bool first_waiting = WaitForWatermarks(*fixture.exporter, 1); + + const auto start = std::chrono::steady_clock::now(); + const bool second_flushed = fixture.exporter->ForceFlush(std::chrono::milliseconds{20}); + const auto ms = std::chrono::duration_cast( std::chrono::steady_clock::now() - start) .count(); slow.join(); + EXPECT_TRUE(first_waiting) << "the first caller never reached its wait, so nothing was queued"; + EXPECT_FALSE(second_flushed); EXPECT_LT(ms, 700) << "waited behind the first caller instead of its own deadline"; } From 4ae3f70fba3bb97b5f6a019228b7417d26628258 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Fri, 14 Aug 2026 22:42:44 +0000 Subject: [PATCH 6/9] Say what the log ordering does not buy Retiring before the terminal diagnostic keeps a flushing log handler from waiting on the session that is reporting itself. It is not general re-entrancy safety, and the comment now says so: a handler calling the unbounded ForceFlush from a progress event blocks an Export that has not handed its request over yet, and one flushing from any client callback can wait on work only that client thread advances. Neither is introduced here and neither is fixed here, they are open telemetry-cpp#4435. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- exporters/elasticsearch/src/es_log_record_exporter.cc | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 07c06fa3cf..fd75573689 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -479,6 +479,13 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( // Logged after the session is retired. The log handler is replaceable, and one that calls // ForceFlush() would otherwise wait for the very session this call has not let go of yet. + // + // That is the whole of what the ordering buys, and it is worth being exact about the rest. + // It does not make a log handler safe to re-enter the exporter from in general: one that + // calls ForceFlush() without a deadline from a progress event blocks the Export() that has + // not handed its request to the client yet, and one that flushes from any callback the HTTP + // client dispatches can wait on work only that client thread can advance. Neither is new + // here and neither is fixed here. if (result != opentelemetry::sdk::common::ExportResult::kSuccess) { OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export " From cf1fd45258b48683561efe7947097792583281c1 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 15 Aug 2026 07:36:21 +0000 Subject: [PATCH 7/9] Let the snapshot wait exist in both configurations The cases that use it are compiled in both on purpose: a case removed from the binary stays registered with CTest and reports a pass without running, so they skip in SetUp instead. Putting the helper they call behind ENABLE_ASYNC_EXPORT broke that, and six call sites failed to compile in a synchronous build. It has a stub there now, inline so that an unused static function does not trip the maintainer mode warning ratchet. Synchronous with maintainer mode builds clean and skips what it should, asynchronous passes 28 of 28, and asynchronous with maintainer mode builds with no warnings. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../elasticsearch/test/es_log_record_exporter_test.cc | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 122c5b16c2..2f009856c4 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -298,6 +298,17 @@ bool WaitForWatermarks(logs_exporter::ElasticsearchLogRecordExporter &exporter, return false; } } // namespace +#else +namespace +{ +// The cases that call this skip in SetUp when there is no wait to observe, but they are compiled +// in both configurations on purpose: a case removed from the binary stays registered with CTest +// and reports a pass without running. So the helper has to exist in both too. +inline bool WaitForWatermarks(logs_exporter::ElasticsearchLogRecordExporter &, std::uint64_t) +{ + return false; +} +} // namespace #endif // ENABLE_ASYNC_EXPORT // --------------------------------------------------------------------------- From f6fa2bbe7346cbdbb142f17e9ea59500ace9ba23 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Mon, 17 Aug 2026 19:11:18 +0000 Subject: [PATCH 8/9] [BUGFIX] Bound the flush, and say nothing on the way to an outcome Eight things from review, each measured rather than accepted on description. The async handler logged five progress states that the synchronous one has always logged and the async one never did. The session is registered before the request is handed to the client, and none of those events retires it, so a log handler calling ForceFlush() from one waited for the export whose call stack it was standing in. Every other report in this file already retires before it logs, which is what made the omission visible: the progress cases were the one place that broke the file's own rule. They report nothing now, and the exhaustive switch stays so a new state is still a compile error. A flush with no deadline of its own waited without one. The bound this path had before the accounting was fixed is response_timeout_, which is also the value the request itself is given, so a session outstanding when the watermark is taken has at most that long before the client owes it a terminal event. The bound is unchanged; what changes is that running out of it no longer reports success. That also keeps this independent of the curl client work, where a transfer can be left with no callback at all. The deadline was built after the mutex was taken, so time spent waiting for the lock came out of nobody's budget and the caller was handed a fresh one. It is taken at entry now, and ForceFlush has a single wait path rather than two. The GiveUpGuard could not reach its active destructor: the shutdown return disarmed it and retired by hand, the successful path disarmed it after SendRequest, and there was no other return between them. It is gone, and the handler takes the completion by move rather than by copy. The invariant it was guarding is stated where the two exits are. WaitForWatermarks had the same thirty second bound as the whole CTest case, so a watermark that never arrived killed the process before the case reached its own assertion. Five seconds for the helper, sixty for CTest. The CMake comment named a case that never parks; it names the one that does. Concurrency between ForceFlush and Shutdown is a MUST in the stable Logs SDK specification and this exporter had no case for it. Two now: one where the shutdown settles the outstanding session and the flush is woken by it rather than by its own bound, and one where the shutdown settles nothing and the flush ends anyway. watermarks_taken is labelled as test synchronization rather than left looking like exporter state. Verified by putting each defect back: a progress line restored fails EverySessionStateIsClassifiedAndReportsAtMostOnce, the bound removed fails ANoDeadlineFlushGivesUpAtTheResponseTimeout, and the unbounded wait restored hangs AParkedFlushEndsEvenIfTheShutdownSettlesNothing at exit 124. All 31 cases pass in the async build, under AddressSanitizer and under ThreadSanitizer, with no sanitizer report and the same count in each. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- exporters/elasticsearch/CMakeLists.txt | 12 +- .../elasticsearch/es_log_record_exporter.h | 11 +- .../src/es_log_record_exporter.cc | 105 ++++++------- .../test/es_log_record_exporter_test.cc | 139 +++++++++++++++++- 4 files changed, 195 insertions(+), 72 deletions(-) diff --git a/exporters/elasticsearch/CMakeLists.txt b/exporters/elasticsearch/CMakeLists.txt index f81248fd26..814fe9aa12 100644 --- a/exporters/elasticsearch/CMakeLists.txt +++ b/exporters/elasticsearch/CMakeLists.txt @@ -56,9 +56,11 @@ if(OTELCPP_BUILD_TESTING) TEST_PREFIX exporter. TEST_LIST es_log_record_exporter_test) - # AnIndefiniteFlushReturnsOnceEverythingIsFinished exercises the branch that - # waits without a deadline, so a regression there does not fail, it stops. - # CTest's default bound is 25 minutes, which is a long time to spend learning - # that. - set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 30) + # AnIndefiniteFlushParksUntilTheOutcomeArrives really does park on the + # condition variable, so a regression there does not fail, it stalls until the + # exporter's own bound expires. CTest's default is 25 minutes, which is a long + # time to spend learning that. This is deliberately well above the bounds the + # cases use themselves, so that one of them expiring reports which case failed + # rather than being killed from the outside. + set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 60) endif() # OTELCPP_BUILD_TESTING diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h index 48d9bc7227..9e97210460 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h @@ -170,11 +170,12 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo std::condition_variable force_flush_cv; std::mutex force_flush_cv_m; - // Raised once per ForceFlush(), under that mutex and immediately after the call has taken - // its watermark. Nothing in the exporter reads it. A case that has to start an export after - // a flush has snapshotted, and before it gives up, waits on this: the alternative is a sleep, - // which leaves the order to the scheduler, and the order that does not reproduce the defect - // is the one that would report a pass. + // Test synchronization only, and not exporter state: nothing here reads it and no behaviour + // depends on it. Raised once per ForceFlush(), under that mutex and immediately after the + // call has taken its watermark. A case that has to start an export after a flush has + // snapshotted, and before it gives up, waits on this. The alternative is a sleep, which + // leaves the order to the scheduler, and the order that does not reproduce the defect is the + // one that would report a pass. std::uint64_t watermarks_taken{0}; }; nostd::shared_ptr synchronization_data_; diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index fd75573689..6bd82309ee 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -338,34 +338,31 @@ class AsyncResponseHandler : public http_client::EventHandler const char *failure = nullptr; switch (state) { + // The states a session passes through on its way to an outcome. Nothing is reported for + // them, and in particular nothing is logged: the log handler is replaceable, this session is + // registered before the request is handed to the client, and none of these events has + // retired it. A handler that flushed from here would wait for the session whose call stack + // it is standing in. Every other report in this file retires first for that reason, and + // there is nothing to retire on the way to an outcome. + 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: failure = "[ES Log Exporter] Create request to elasticsearch failed"; break; - case http_client::SessionState::Created: - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session created"); - break; case http_client::SessionState::Destroyed: failure = "[ES Log Exporter] Session to elasticsearch destroyed before a response"; break; - case http_client::SessionState::Connecting: - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connecting to elasticsearch"); - break; case http_client::SessionState::ConnectFailed: failure = "[ES Log Exporter] Connection to elasticsearch failed"; break; - case http_client::SessionState::Connected: - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connected to elasticsearch"); - break; - case http_client::SessionState::Sending: - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Sending request to elasticsearch"); - break; case http_client::SessionState::SendFailed: failure = "[ES Log Exporter] Request failed to be sent to elasticsearch"; break; - case http_client::SessionState::Response: - // The body arrives through OnResponse(), which is what reports the outcome. - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Response received from elasticsearch"); - break; case http_client::SessionState::SSLHandshakeFailed: failure = "[ES Log Exporter] SSL handshake to elasticsearch failed"; break; @@ -482,10 +479,9 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( // // That is the whole of what the ordering buys, and it is worth being exact about the rest. // It does not make a log handler safe to re-enter the exporter from in general: one that - // calls ForceFlush() without a deadline from a progress event blocks the Export() that has - // not handed its request to the client yet, and one that flushes from any callback the HTTP - // client dispatches can wait on work only that client thread can advance. Neither is new - // here and neither is fixed here. + // flushes from any callback the HTTP client dispatches can still wait on work that only the + // client thread it is standing on can advance. That is + // https://github.com/open-telemetry/opentelemetry-cpp/issues/4435, and it is not fixed here. if (result != opentelemetry::sdk::common::ExportResult::kSuccess) { OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export " @@ -500,37 +496,19 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( return true; }; - // A return added between here and SendRequest() would otherwise strand a waiter on a session - // that can never finish. Reporting through the same completion leaves a session one way out, - // and says so in the log rather than dropping the batch silently. - struct GiveUpGuard - { - Completion *report = nullptr; - - GiveUpGuard() = default; - GiveUpGuard(const GiveUpGuard &) = delete; - GiveUpGuard &operator=(const GiveUpGuard &) = delete; - GiveUpGuard(GiveUpGuard &&) = delete; - GiveUpGuard &operator=(GiveUpGuard &&) = delete; - - ~GiveUpGuard() - { - if (report != nullptr) - { - (*report)(opentelemetry::sdk::common::ExportResult::kFailure); - } - } - } guard; - guard.report = &complete; + // Two ways out from here, and each one is spelled out below: the shutdown refusal retires by + // hand, and the handler takes the session over once SendRequest() has it. A return added + // between them has to do one or the other, or it strands a waiter on a session that can never + // finish. #endif // Return failure if this exporter has been shutdown if (isShutdown()) { #ifdef ENABLE_ASYNC_EXPORT - // Retired before anything replaceable runs, and reported here rather than through the guard, - // so a log handler that flushes does not wait for this Export() and one refusal reads as one. - guard.report = nullptr; + // Retired before anything replaceable runs, and reported by the line below rather than + // through the completion, so a log handler that flushes does not wait for this Export() and + // one refusal reads as one. retire(); #endif OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting " @@ -573,10 +551,11 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( #ifdef ENABLE_ASYNC_EXPORT // Send the request - auto handler = std::make_shared(session, Completion(complete), - options_.console_debug_); + // The handler reports this session from here on, so the completion goes to it rather than + // being copied to it. + auto handler = + std::make_shared(session, std::move(complete), options_.console_debug_); session->SendRequest(handler); - guard.report = nullptr; // the handler reports this session from here on return sdk::common::ExportResult::kSuccess; #else // Send the request @@ -627,6 +606,25 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout( timeout, std::chrono::microseconds::zero()); + // A caller that asks for no deadline still gets one, because a flush that cannot end is worse + // than one that reports it did not finish, and the client is not obliged to call back at all. + // response_timeout_ is the bound this wait already had before the accounting was fixed, and it + // is the same value the request itself is given, so a session outstanding when the watermark is + // taken has at most that long before the client owes it a terminal event. The bound is + // unchanged; what changes is that running out of it no longer reports success. + if (timeout <= std::chrono::microseconds::zero()) + { + timeout = std::chrono::seconds{options_.response_timeout_}; + } + + // Taken before the lock rather than after it, so the time spent waiting for the mutex comes out + // of the caller's budget instead of being handed back as a fresh one. A plain mutex still + // cannot promise a bound, but acquiring it should not add a second full timeout to the one that + // was asked for. + const std::chrono::steady_clock::time_point deadline = + std::chrono::steady_clock::now() + + std::chrono::duration_cast(timeout); + std::unique_lock lock(synchronization_data_->force_flush_cv_m); // The snapshot is the next id, not a count: a session started after it takes a larger id and @@ -639,20 +637,9 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou return running.empty() || *running.begin() >= watermark; }; - if (timeout <= std::chrono::microseconds::zero()) - { - // wait() only returns once the predicate holds, so the flush has completed. - synchronization_data_->force_flush_cv.wait(lock, flushed); - return true; - } - // One deadline for the call, so a wakeup that is not a completion resumes against what is left // rather than starting the wait again. wait_until() returns the predicate, so a flush that ran // out of time cannot report success. - const std::chrono::steady_clock::time_point deadline = - std::chrono::steady_clock::now() + - std::chrono::duration_cast(timeout); - return synchronization_data_->force_flush_cv.wait_until(lock, deadline, flushed); #else return true; diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 2f009856c4..12a04c7ec7 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -244,7 +244,20 @@ class FakeHttpClient : public http_client::HttpClient // Runs inside Export(), after the records have been handed over and before the request exists. std::function on_create_session; - bool CancelAllSessions() noexcept override { return true; } + + // 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 {} @@ -283,10 +296,15 @@ namespace { // Bounded, and the count only goes up, so a wait that expires means the flush never got there // rather than that the case missed it. +// +// Well inside the bound CTest puts on the whole binary. Matching the two would mean a case that +// never takes its watermark is killed from the outside before this returns, so it would never +// reach its own assertion, its join, or its cleanup, and the report would say the suite timed out +// rather than which case failed and why. bool WaitForWatermarks(logs_exporter::ElasticsearchLogRecordExporter &exporter, std::uint64_t wanted) { - const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds{30}; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds{5}; while (std::chrono::steady_clock::now() < deadline) { if (logs_exporter::ElasticsearchExporterTestPeer::WatermarksTaken(exporter) >= wanted) @@ -718,6 +736,98 @@ TEST_F(ElasticsearchForceFlushTests, AFailedExportSettlesItsSessionAndIsReported EXPECT_TRUE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20})); } + +// A caller with no deadline of its own still gets one. Nothing obliges an HTTP client to call +// back, and a flush that waits for a callback that never comes would take the caller down with +// it, so the exporter bounds the wait by its own response timeout and reports that it did not +// finish. The elapsed time is asserted from below as well: a bound of zero would also return +// false here, and would return it immediately. +TEST_F(ElasticsearchForceFlushTests, ANoDeadlineFlushGivesUpAtTheResponseTimeout) +{ + // Kept for the same reason as the case at the top of this file: a dropped handler reports a + // failure from its destructor, which would retire the session this case needs left outstanding. + std::vector> kept; + auto fixture = MakeExporter([&kept](const std::shared_ptr &handler) { + kept.push_back(handler); + }); + ExportOnce(*fixture.exporter); + + const auto start = std::chrono::steady_clock::now(); + const bool flushed = fixture.exporter->ForceFlush(); + const auto elapsed = std::chrono::steady_clock::now() - start; + const auto ms = std::chrono::duration_cast(elapsed).count(); + + EXPECT_FALSE(flushed) << "reported a completion for a session that never answered"; + EXPECT_GE(ms, 1000) << "returned without waiting, so the bound is not the response timeout"; +} + +// ForceFlush() and Shutdown() are required to be safe to call concurrently, and this PR took out +// the lock that used to serialise them, so the overlap is deterministic here rather than left to +// chance: the flush is parked on its session before the shutdown starts. A real client answers +// its outstanding sessions while shutting down, which is what has to wake the waiter. +TEST_F(ElasticsearchForceFlushTests, AParkedFlushIsWokenByTheShutdownThatCompletesItsSession) +{ + std::shared_ptr captured; + auto fixture = + MakeExporter([&captured](const std::shared_ptr &handler) { + captured = handler; + }); + ExportOnce(*fixture.exporter); + ASSERT_NE(captured, nullptr); + + fixture.client->on_cancel_all = [&captured] { + captured->OnEvent(http_client::SessionState::Cancelled, ""); + }; + + std::atomic returned{false}; + bool flushed = false; + std::thread waiter([&fixture, &returned, &flushed] { + flushed = fixture.exporter->ForceFlush(); + returned.store(true, std::memory_order_release); + }); + + // Recorded rather than asserted, for the reason the cases above give: nothing fatal may happen + // while that thread is still joinable. + const bool snapshotted = WaitForWatermarks(*fixture.exporter, 1); + const bool returned_early = returned.load(std::memory_order_acquire); + + const auto start = std::chrono::steady_clock::now(); + const bool down = fixture.exporter->Shutdown(); + waiter.join(); + const auto ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); + + EXPECT_TRUE(snapshotted) << "the flush never took a watermark, so it never reached the wait"; + EXPECT_FALSE(returned_early) << "the flush returned with its session still outstanding"; + EXPECT_TRUE(down); + EXPECT_TRUE(flushed) << "the shutdown settled the session and the flush still reported failure"; + EXPECT_LT(ms, kShortResponseTimeoutSeconds * 1000) + << "the flush waited out its own bound instead of being woken by the shutdown"; +} + +// The same overlap against a client that answers nothing on the way down. There is no event to +// wake the waiter here, so what is being pinned is that the flush still ends: without a bound on +// the no-deadline wait this case does not fail, it stops. +TEST_F(ElasticsearchForceFlushTests, AParkedFlushEndsEvenIfTheShutdownSettlesNothing) +{ + std::vector> kept; + auto fixture = MakeExporter([&kept](const std::shared_ptr &handler) { + kept.push_back(handler); + }); + ExportOnce(*fixture.exporter); + + bool flushed = false; + std::thread waiter([&fixture, &flushed] { flushed = fixture.exporter->ForceFlush(); }); + + const bool snapshotted = WaitForWatermarks(*fixture.exporter, 1); + const bool down = fixture.exporter->Shutdown(); + waiter.join(); + + EXPECT_TRUE(snapshotted) << "the flush never took a watermark, so it never reached the wait"; + EXPECT_TRUE(down); + EXPECT_FALSE(flushed) << "reported a completion for a session nothing ever settled"; +} // --------------------------------------------------------------------------- // Exactly-once accounting for the async handler, which exists only in an async build, so // these cases skip there rather than compile out. @@ -743,6 +853,8 @@ class CompletionCountingLogHandler : public internal_log::LogHandler { return; } + lines_.fetch_add(1, std::memory_order_relaxed); + const std::string text(msg); if (text.find("log record(s) success") != std::string::npos) { @@ -758,9 +870,14 @@ 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. + 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 @@ -798,6 +915,7 @@ class ElasticsearchAsyncCompletionTests : public ::testing::Test } int Completions() const { return Counter().completions(); } + int Lines() const { return Counter().lines(); } nostd::shared_ptr handler_; nostd::shared_ptr previous_handler_; @@ -848,11 +966,26 @@ TEST_F(ElasticsearchAsyncCompletionTests, EverySessionStateIsClassifiedAndReport // Counted per iteration: the previous fixture's handler reports from its destructor as it goes // out of scope, which lands in the same counter. - const int before = Completions(); + const int before = Completions(); + const int lines_before = Lines(); ExportOnce(*fixture.exporter); EXPECT_EQ(Completions() - before, test_case.terminal ? 1 : 0); EXPECT_EQ(fixture.exporter->ForceFlush(std::chrono::milliseconds{20}), test_case.terminal); + + // Nothing on the way to an outcome may reach the log handler at all. The handler is + // replaceable, this session is registered before the request is handed to the client, and no + // progress event has retired it, so a handler that flushed from one would be waiting for the + // export whose call stack it is standing in. A terminal state is reported, and by then the + // session has already been let go. + if (test_case.terminal) + { + EXPECT_GT(Lines() - lines_before, 0) << "a terminal state said nothing"; + } + else + { + EXPECT_EQ(Lines() - lines_before, 0) << "a progress event reached the log handler"; + } } } From 0eb3c55848e986021531be28ef70ce45a7b7328d Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Tue, 25 Aug 2026 14:20:32 +0000 Subject: [PATCH 9/9] [CHORE] Say what the code does rather than what it stopped doing Three comments described the change relative to the previous shape: a bound that is "unchanged", a lock that was taken out, and an ordering that "no longer" comes from the curl client. A reader of the file has no previous shape to compare against, and the rationale for the change is in this PR. Each now states the current fact: what response_timeout_ bounds, that nothing serialises ForceFlush against Shutdown, and that the curl client does not produce the ordering the case pins while another client may. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../elasticsearch/src/es_log_record_exporter.cc | 10 ++++------ .../test/es_log_record_exporter_test.cc | 14 +++++++------- 2 files changed, 11 insertions(+), 13 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 6bd82309ee..e3f5fb8638 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -606,12 +606,10 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout( timeout, std::chrono::microseconds::zero()); - // A caller that asks for no deadline still gets one, because a flush that cannot end is worse - // than one that reports it did not finish, and the client is not obliged to call back at all. - // response_timeout_ is the bound this wait already had before the accounting was fixed, and it - // is the same value the request itself is given, so a session outstanding when the watermark is - // taken has at most that long before the client owes it a terminal event. The bound is - // unchanged; what changes is that running out of it no longer reports success. + // A caller that asks for no deadline still gets one: a flush that cannot end is worse than one + // that reports it did not finish, and a client is not obliged to call back at all. + // response_timeout_ is the same value the request itself is given, so a session outstanding when + // the watermark is taken has at most that long before the client owes it a terminal event. if (timeout <= std::chrono::microseconds::zero()) { timeout = std::chrono::seconds{options_.response_timeout_}; diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 12a04c7ec7..9854477687 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -761,10 +761,10 @@ TEST_F(ElasticsearchForceFlushTests, ANoDeadlineFlushGivesUpAtTheResponseTimeout EXPECT_GE(ms, 1000) << "returned without waiting, so the bound is not the response timeout"; } -// ForceFlush() and Shutdown() are required to be safe to call concurrently, and this PR took out -// the lock that used to serialise them, so the overlap is deterministic here rather than left to -// chance: the flush is parked on its session before the shutdown starts. A real client answers -// its outstanding sessions while shutting down, which is what has to wake the waiter. +// ForceFlush() and Shutdown() have to be safe to call concurrently and nothing serialises them, +// so the overlap here is arranged rather than left to chance: the flush is parked on its session +// before the shutdown starts. A real client answers its outstanding sessions while shutting down, +// which is what has to wake the waiter. TEST_F(ElasticsearchForceFlushTests, AParkedFlushIsWokenByTheShutdownThatCompletesItsSession) { std::shared_ptr captured; @@ -1315,9 +1315,9 @@ TEST_F(ElasticsearchAsyncCompletionTests, TwoTerminalEventsCountAsOneSession) << "the second session is still in flight"; } -// The curl client's completion lambda tests the response first and the abort in an else, so this -// ordering no longer comes from it. It still comes from anywhere: EventHandler promises callers -// nothing about how many terminal events arrive, and #4360 is open on exactly that. +// The curl client's completion lambda tests the response first and the abort in an else, so it +// does not produce this ordering. Another client may: EventHandler promises callers nothing about +// how many terminal events arrive, and #4360 is open on exactly that. TEST_F(ElasticsearchAsyncCompletionTests, CancelledThenAResponseCountsOnce) { std::vector> kept;