diff --git a/CHANGELOG.md b/CHANGELOG.md index 4697267017..eef3aa5a74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,9 @@ Increment the: request body on a 32-bit build [#4630](https://github.com/open-telemetry/opentelemetry-cpp/pull/4630) +* [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); + } +}