diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ff05e36d9..048c4e0cba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,9 @@ Increment the: * [OTLP/HTTP] Honor `Retry-After` response header when retrying exports, supporting both delay-seconds and HTTP-date formats per RFC 7231 ยง7.1.3. [#4172](https://github.com/open-telemetry/opentelemetry-cpp/issues/4172) + +* [BUG] End the Elasticsearch exporter's wait on a read or write error + [#4331](https://github.com/open-telemetry/opentelemetry-cpp/pull/4331) * [BUILD] Run the ext_http component install test on Windows [#4326](https://github.com/open-telemetry/opentelemetry-cpp/pull/4326) diff --git a/exporters/elasticsearch/CMakeLists.txt b/exporters/elasticsearch/CMakeLists.txt index cfd4ce01db..4ce3a11f85 100644 --- a/exporters/elasticsearch/CMakeLists.txt +++ b/exporters/elasticsearch/CMakeLists.txt @@ -55,4 +55,8 @@ if(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() # BUILD_TESTING diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index a87a514f01..885b3c5cd3 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -108,8 +108,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() { @@ -182,10 +182,19 @@ class ResponseHandler : public http_client::EventHandler recordCompletion(CompletionState::Failure); break; case http_client::SessionState::ReadError: - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Read error"); + // Reported only when this is the outcome Export() returns. Recording is first writer + // wins, so either of these can arrive after a response has already succeeded, and an + // error line there would describe a failure the caller was never told about. + if (recordCompletion(CompletionState::Failure)) + { + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Read error"); + } break; case http_client::SessionState::WriteError: - OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Write error"); + if (recordCompletion(CompletionState::Failure)) + { + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Write error"); + } break; case http_client::SessionState::Cancelled: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] (manually) cancelled"); @@ -206,22 +215,27 @@ class ResponseHandler : public http_client::EventHandler * Record the outcome of the request, first writer wins, then release any waiter. Keeping the * first outcome means a session destroyed after a successful response does not overwrite it. */ - void recordCompletion(CompletionState state) + /// Returns whether this call is the one that decided the outcome. + bool recordCompletion(CompletionState state) { + bool recorded = false; { std::unique_lock lk(mutex_); - recordCompletionLocked(state); + recorded = recordCompletionLocked(state); } cv_.notify_all(); + return recorded; } /// As recordCompletion(), for callers that already hold mutex_ and notify themselves. - void recordCompletionLocked(CompletionState state) + bool recordCompletionLocked(CompletionState state) { - if (completion_ == CompletionState::Pending) + if (completion_ != CompletionState::Pending) { - completion_ = state; + return false; } + completion_ = state; + return true; } // Define a condition variable and mutex diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index a65c0b4c1c..216b724d07 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -4,7 +4,9 @@ #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" @@ -19,6 +21,8 @@ #include #include #include +#include +#include #include #include #include "nlohmann/json.hpp" @@ -142,3 +146,245 @@ TEST(ElasticsearchLogRecordableTests, BasicTests) EXPECT_EQ(actual, expected); } + +// --------------------------------------------------------------------------- +// Synchronous completion path. +// +// A fake HTTP client drives a scripted sequence of callbacks from inside +// SendRequest(), which runs before the exporter reaches waitForResponse(). Every +// case here therefore also covers a completion recorded before the wait starts, +// the notification a bare cv_.wait() would have missed. +// --------------------------------------------------------------------------- +namespace +{ +namespace http_client = opentelemetry::ext::http::client; + +// Accepted by both the substring check and a top level "errors": false parse, 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":{"_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 + { + 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_; +}; + +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 + +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. The seven states other than the two +// this change adds have behaved this way since #4298 but had no test. +// +// 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); +} + +// Same rule for the two states this change makes terminal, and the reason their error line is +// conditional: reaching one of them says nothing on its own about the result Export() reports. +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, and the case where the error line is the only diagnostic the caller gets: +// 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); + } +}