From f71ea10b4c8dd112906ec87b5bfc9964c7f61436 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 19 Sep 2026 14:59:39 +0000 Subject: [PATCH] [BUG] End the Elasticsearch exporter's wait on a read or write error ResponseHandler::OnEvent records a completion for each session state that ends the session, but ReadError and WriteError only log. A client that reports either of them as its last event leaves the synchronous Export() blocked on its condition variable with nothing left to wake it. The bundled curl client dispatches neither state, so this does not change behaviour for callers who use it. It matters for the HttpClient that the exporter's constructor accepts. The cases live in the existing synchronous fixture, which skips when the exporter is built for async export, and a CTest timeout bounds them because a regression here stalls the job rather than failing an assertion. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 3 + exporters/elasticsearch/CMakeLists.txt | 4 + .../src/es_log_record_exporter.cc | 8 +- .../test/es_log_record_exporter_test.cc | 350 +++++++++++++++++- 4 files changed, 352 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d1766079f9..b03ad5f7ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,9 @@ Increment the: to compile standalone on newer standard library implementations. [#4574](https://github.com/open-telemetry/opentelemetry-cpp/pull/4574) +* [BUG] End the Elasticsearch exporter's wait on a read or write error + [#4331](https://github.com/open-telemetry/opentelemetry-cpp/pull/4331) + ## [1.29.0] 2026-09-13 * [RELEASE] Bump main branch to 1.29.0-dev (#4259) diff --git a/exporters/elasticsearch/CMakeLists.txt b/exporters/elasticsearch/CMakeLists.txt index 18b2bcd898..9f1990049c 100644 --- a/exporters/elasticsearch/CMakeLists.txt +++ b/exporters/elasticsearch/CMakeLists.txt @@ -55,4 +55,8 @@ if(OTELCPP_BUILD_TESTING) TARGET es_log_record_exporter_test TEST_PREFIX exporter. TEST_LIST es_log_record_exporter_test) + + # These cases exist to catch a wait that never returns. Without a per test + # bound a regression stalls the job instead of failing it. + set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 30) endif() # OTELCPP_BUILD_TESTING diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index cd3b1bdbfd..89e90f352c 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -110,8 +110,8 @@ class ResponseHandler : public http_client::EventHandler /** * A method the user calls to block their thread until the request has either produced a - * response or failed. The longest duration is the timeout of the request, set by - * SetTimeoutMs(), which arrives here as a TimedOut session event. + * response or failed. It has no deadline of its own and relies on the HTTP client + * reporting one of the terminal session states. */ bool waitForResponse() { @@ -183,11 +183,15 @@ class ResponseHandler : public http_client::EventHandler OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Network error"); recordCompletion(CompletionState::Failure); break; + // Both of these end the session, so each has to release the waiter. Without that a + // synchronous export whose transfer fails this way blocks until the process ends. case http_client::SessionState::ReadError: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Read error"); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::WriteError: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Write error"); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::Cancelled: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] (manually) cancelled"); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 52a5d9b6ef..27c8b6e6da 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -21,7 +21,11 @@ #include #include #include +#include +#include +#include #include +#include #include #include "nlohmann/json.hpp" @@ -32,14 +36,15 @@ namespace http_client = opentelemetry::ext::http::client; // A response shaped like a successful Elasticsearch bulk reply: the exporter looks for // `"failed" : 0` in the body (see ElasticsearchLogRecordExporter::Export) in addition to the // status code before reporting success. +constexpr const char *kDefaultAcceptedBody = R"({"errors": false, "failed" : 0})"; + class FakeResponse final : public http_client::Response { public: - FakeResponse() - { - static const std::string kSuccessBody = R"({"errors": false, "failed" : 0})"; - body_.assign(kSuccessBody.begin(), kSuccessBody.end()); - } + explicit FakeResponse(http_client::StatusCode status = 200, + const std::string &body = kDefaultAcceptedBody) + : status_(status), body_(body.begin(), body.end()) + {} const http_client::Body &GetBody() const noexcept override { return body_; } @@ -58,9 +63,10 @@ class FakeResponse final : public http_client::Response return true; } - http_client::StatusCode GetStatusCode() const noexcept override { return 200; } + http_client::StatusCode GetStatusCode() const noexcept override { return status_; } private: + http_client::StatusCode status_; http_client::Body body_; }; @@ -85,11 +91,24 @@ class FakeRequest final : public http_client::Request void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {} }; -// A session whose SendRequest() answers synchronously with a successful FakeResponse, so the -// exporter's own wait for a response returns immediately without needing a real connection. +// What the client does with a request, called from inside SendRequest() so the exporter's own +// wait returns without needing a connection. The default answers once, successfully, which is +// what a case wants when the response is not the thing under test. +using EventScript = std::function; + +EventScript AnswerSuccessfully() +{ + return [](http_client::EventHandler &handler) { + FakeResponse response; + handler.OnResponse(response); + }; +} + class FakeSession final : public http_client::Session { public: + explicit FakeSession(EventScript script = AnswerSuccessfully()) : script_(std::move(script)) {} + std::shared_ptr CreateRequest() noexcept override { return std::make_shared(); @@ -97,27 +116,34 @@ class FakeSession final : public http_client::Session void SendRequest(std::shared_ptr handler) noexcept override { - FakeResponse response; - handler->OnResponse(response); + script_(*handler); } bool IsSessionActive() noexcept override { return true; } bool CancelSession() noexcept override { return true; } bool FinishSession() noexcept override { return true; } + +private: + EventScript script_; }; class FakeHttpClient final : public http_client::HttpClient { public: + explicit FakeHttpClient(EventScript script = AnswerSuccessfully()) : script_(std::move(script)) {} + std::shared_ptr CreateSession( opentelemetry::nostd::string_view) noexcept override { - return std::make_shared(); + return std::make_shared(script_); } bool CancelAllSessions() noexcept override { return true; } bool FinishAllSessions() noexcept override { return true; } void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + EventScript script_; }; } // namespace @@ -263,3 +289,305 @@ TEST(ElasticsearchLogRecordableTests, BasicTests) EXPECT_EQ(actual, expected); } + +// Synchronous completion path. The fake client scripts its callbacks from inside SendRequest(), +// which runs before the exporter reaches waitForResponse(), so every case here also covers a +// completion recorded before the wait starts. +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":{"_index":"logs","status":201,"_shards":{"failed" : 0}}}]})"; + +// Keeps the handler and returns, so the export reaches its wait with nothing recorded and only a +// notification can end it. +class DeferredSession : public http_client::Session +{ +public: + explicit DeferredSession(std::promise> *arrived) + : arrived_(arrived) + {} + + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + // The handler travels in the promise rather than beside it. A waiter that times out has not + // observed the promise becoming ready and so is not synchronized with this thread, which would + // make a handler read on that path a race with this write. + void SendRequest(std::shared_ptr handler) noexcept override + { + arrived_->set_value(std::move(handler)); + } + bool IsSessionActive() noexcept override { return true; } + bool CancelSession() noexcept override { return true; } + bool FinishSession() noexcept override { return true; } + +private: + std::promise> *arrived_; +}; + +class DeferredHttpClient : public http_client::HttpClient +{ +public: + explicit DeferredHttpClient(std::promise> *arrived) + : arrived_(arrived) + {} + + std::shared_ptr CreateSession(nostd::string_view) noexcept override + { + return std::make_shared(arrived_); + } + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + std::promise> *arrived_; +}; + +opentelemetry::sdk::common::ExportResult ExportWith(EventScript script) +{ + auto client = std::make_shared(std::move(script)); + logs_exporter::ElasticsearchExporterOptions options; + logs_exporter::ElasticsearchLogRecordExporter exporter(options, client); + auto record = exporter.MakeRecordable(); + return exporter.Export(nostd::span>(&record, 1)); +} +} // namespace + +// The synchronous wait exists only when the exporter is built without async export, so these cases +// skip rather than compile out: gtest_add_tests reads the source, and a case that disappeared from +// the binary would still be registered with CTest. The skip goes in SetUp rather than at the top of +// each body, because GTEST_SKIP returns and leaves the rest of the body unreachable, which MSVC +// reports as C4702 and the maintainer mode jobs turn into an error. +namespace +{ +class ElasticsearchLogsExporterSyncTests : public ::testing::Test +{ +protected: + void SetUp() override + { +#ifdef ENABLE_ASYNC_EXPORT + GTEST_SKIP() << "Export() returns without waiting when async export is enabled"; +#endif + } +}; +} // namespace + +namespace +{ +// Runs one export on another thread against a session that keeps its handler, and hands back the +// handler once the exporter has reached it. Everything is held by shared_ptr and the thread is +// detached, because waitForResponse has no deadline: if a wake-up stops working the export never +// returns, and joining it would hang the binary rather than fail the case. +struct ParkedExport +{ + std::shared_ptr client; + std::shared_ptr exporter; + std::shared_ptr handler; + std::promise> arrived; + std::promise done; + std::future finished; +}; + +// Answers with nullptr rather than a half prepared handle when a precondition does not hold. Each +// step below is something a case needs before it can mean anything, and carrying on past one of +// them reads state the other thread is still writing. +std::shared_ptr StartParkedExport() +{ + auto parked = std::make_shared(); + parked->client = std::make_shared(&parked->arrived); + + logs_exporter::ElasticsearchExporterOptions options; + parked->exporter = + std::make_shared(options, parked->client); + + auto reached = parked->arrived.get_future(); + auto finished = parked->done.get_future(); + + std::thread([parked] { + auto record = parked->exporter->MakeRecordable(); + parked->done.set_value( + parked->exporter->Export(nostd::span>(&record, 1))); + }).detach(); + + if (std::future_status::ready != reached.wait_for(std::chrono::seconds{5})) + { + ADD_FAILURE() << "the exporter never handed off the request handler"; + return nullptr; + } + + parked->handler = reached.get(); + if (!parked->handler) + { + ADD_FAILURE() << "the session was handed a null handler"; + return nullptr; + } + + if (std::future_status::timeout != finished.wait_for(std::chrono::milliseconds{100})) + { + ADD_FAILURE() << "the export returned before any callback was delivered"; + return nullptr; + } + + parked->finished = std::move(finished); + return parked; +} +} // namespace + +// A terminal error that arrives after the handler has been handed off, while the export is still +// running, has to end the wait, which is the half the notification is responsible for. Without +// these two, removing cv_.notify_all() keeps this file green. +// +// The handoff is what they hold, not the parking: the session publishes the handler before +// SendRequest() returns, so the export need not have reached cv_.wait() when the callback is +// delivered. Holding that would want a wait entry seam in production code, which is not worth the +// API it would add. +TEST_F(ElasticsearchLogsExporterSyncTests, AReadErrorAfterTheHandoffEndsTheExport) +{ + auto parked = StartParkedExport(); + ASSERT_NE(nullptr, parked); + + parked->handler->OnEvent(http_client::SessionState::ReadError, ""); + + ASSERT_EQ(std::future_status::ready, parked->finished.wait_for(std::chrono::seconds{10})) + << "the read error never woke the export"; + EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kFailure, parked->finished.get()); +} + +// The same for the success half, which also holds that the wait is a wait: a waitForResponse that +// only read the current state would answer before this response arrives. +TEST_F(ElasticsearchLogsExporterSyncTests, AResponseAfterTheHandoffEndsTheExport) +{ + auto parked = StartParkedExport(); + ASSERT_NE(nullptr, parked); + + FakeResponse response(200, kAcceptedBody); + parked->handler->OnResponse(response); + + ASSERT_EQ(std::future_status::ready, parked->finished.wait_for(std::chrono::seconds{10})) + << "the response never woke the export"; + EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kSuccess, parked->finished.get()); +} + +TEST_F(ElasticsearchLogsExporterSyncTests, ResponseRecordedBeforeTheWaitIsStillSeen) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + FakeResponse response(200, kAcceptedBody); + handler.OnResponse(response); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); +} + +TEST_F(ElasticsearchLogsExporterSyncTests, ReadErrorEndsTheWait) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + handler.OnEvent(http_client::SessionState::ReadError, ""); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); +} + +// The whole contract in one place. Every state that ends a session has to leave a result behind, +// otherwise a client that emits it last strands the wait. +// +// A regression here surfaces as a CTest timeout rather than a failed assertion, because a state +// that stops being terminal leaves Export() waiting with nothing left to wake it. +TEST_F(ElasticsearchLogsExporterSyncTests, EveryTerminalStateEndsTheWaitInFailure) +{ + const http_client::SessionState terminal[] = { + http_client::SessionState::CreateFailed, http_client::SessionState::ConnectFailed, + http_client::SessionState::SendFailed, http_client::SessionState::SSLHandshakeFailed, + http_client::SessionState::TimedOut, http_client::SessionState::NetworkError, + http_client::SessionState::Cancelled, http_client::SessionState::ReadError, + http_client::SessionState::WriteError, http_client::SessionState::Destroyed}; + + for (const auto state : terminal) + { + SCOPED_TRACE(static_cast(state)); + const auto result = + ExportWith([state](http_client::EventHandler &handler) { handler.OnEvent(state, ""); }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); + } +} + +// The other side of the same contract: a state that only reports progress must not complete the +// export on its own, or a response that arrives afterwards is never consulted. +TEST_F(ElasticsearchLogsExporterSyncTests, ProgressStatesDoNotDecideTheResult) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + handler.OnEvent(http_client::SessionState::Created, ""); + handler.OnEvent(http_client::SessionState::Connecting, ""); + handler.OnEvent(http_client::SessionState::Connected, ""); + handler.OnEvent(http_client::SessionState::Sending, ""); + handler.OnEvent(http_client::SessionState::Response, ""); + FakeResponse response(200, kAcceptedBody); + handler.OnResponse(response); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); +} + +TEST_F(ElasticsearchLogsExporterSyncTests, WriteErrorEndsTheWait) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + handler.OnEvent(http_client::SessionState::WriteError, ""); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); +} + +TEST_F(ElasticsearchLogsExporterSyncTests, SessionDestroyedWhilePendingEndsTheWait) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + handler.OnEvent(http_client::SessionState::Destroyed, ""); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); +} + +// The first outcome recorded is the one reported, so tearing the session down after a response has +// arrived does not turn a successful export into a failure. +TEST_F(ElasticsearchLogsExporterSyncTests, SessionDestroyedAfterAResponseKeepsTheSuccess) +{ + const auto result = ExportWith([](http_client::EventHandler &handler) { + FakeResponse response(200, kAcceptedBody); + handler.OnResponse(response); + handler.OnEvent(http_client::SessionState::Destroyed, ""); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); +} + +// The first outcome still wins once these two are terminal: reaching an I/O error after a +// response has been recorded must not turn a successful export into a failure. +TEST_F(ElasticsearchLogsExporterSyncTests, IoErrorAfterAResponseKeepsTheSuccess) +{ + for (const auto state : + {http_client::SessionState::ReadError, http_client::SessionState::WriteError}) + { + const auto result = ExportWith([state](http_client::EventHandler &handler) { + FakeResponse response(200, kAcceptedBody); + handler.OnResponse(response); + handler.OnEvent(state, ""); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); + } +} + +// The mirror image: the I/O error is recorded first, so a response arriving afterwards does +// not rescue the export. +TEST_F(ElasticsearchLogsExporterSyncTests, IoErrorBeforeAResponseKeepsTheFailure) +{ + for (const auto state : + {http_client::SessionState::ReadError, http_client::SessionState::WriteError}) + { + SCOPED_TRACE(static_cast(state)); + const auto result = ExportWith([state](http_client::EventHandler &handler) { + handler.OnEvent(state, ""); + FakeResponse response(200, kAcceptedBody); + handler.OnResponse(response); + }); + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); + } +}