diff --git a/CHANGELOG.md b/CHANGELOG.md index 098a96cc33..1b85981b18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,19 @@ Increment the: ## [Unreleased] +* [EXPORTER] Fix the Elasticsearch log exporter's synchronous export path + waiting with no deadline of its own, trusting an injected `HttpClient` to + always eventually deliver a terminal event. A client that accepts a + request and never calls back (a dead thread, a reused socket, a swallowed + error) left `Export()` blocked for the life of the process. The wait now + has its own deadline derived from the configured response timeout, so a + non-responding client fails the export instead of hanging it. A + `response_timeout` of 0 means no timeout, matching the curl client's own + `CURLOPT_TIMEOUT_MS` 0 semantics, so it no longer maps to a deadline + already in the past: that previously failed every synchronous export + immediately, even one the client was about to answer successfully. + [#4362](https://github.com/open-telemetry/opentelemetry-cpp/issues/4362) + ## [1.30.0] 2026-10-22 This release contains: diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index cd3b1bdbfd..48560b35d9 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -110,15 +110,34 @@ 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, or until the given deadline passes. Ordinarily the request's own + * timeout (set by SetTimeoutMs()) arrives here first; the curl client reports that as a + * SendFailed session event, the same as any other transport failure, since it does not + * distinguish a timeout from other CURL errors. But a terminal event at all is a guarantee + * that belongs to the injected HttpClient, not to this exporter: a client that accepts a + * handler and never delivers one (a dead thread, a reused socket, a swallowed error) would + * otherwise leave this wait blocked for the life of the process. The deadline is + * this exporter's own backstop, independent of whether the client honors its side of the + * contract. + * + * @param timed_out if not null, set to whether the deadline passed with nothing having + * reaped the transfer yet (completion_ still Pending), as opposed to a terminal event (a + * response, or a failure like ConnectFailed/SendFailed) having already arrived. The caller + * needs this distinction: only a still-outstanding transfer needs CancelSession() rather than + * FinishSession(), since a terminal event means the transfer is already over. */ - bool waitForResponse() + bool waitForResponse(std::chrono::steady_clock::time_point deadline, bool *timed_out = nullptr) { std::unique_lock lk(mutex_); // Waiting on a predicate rather than bare: the completion may already have been recorded // before this thread got here, in which case there is no notification left to receive. - cv_.wait(lk, [this] { return completion_ != CompletionState::Pending; }); + // A deadline that passes without a terminal event leaves completion_ at Pending, which + // reads as failure below, the same outcome a terminal error event would have produced. + cv_.wait_until(lk, deadline, [this] { return completion_ != CompletionState::Pending; }); + if (timed_out != nullptr) + { + *timed_out = (completion_ == CompletionState::Pending); + } return completion_ == CompletionState::Success; } @@ -470,6 +489,15 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( #else // Send the request auto handler = std::make_shared(options_.console_debug_); + // Captured before SendRequest() so the deadline reflects this exporter's own timeout budget, + // not whatever the injected HttpClient decides to do with it (see waitForResponse()). A + // non-positive response_timeout_ means no timeout, matching the curl client's own + // CURLOPT_TIMEOUT_MS 0 semantics, so it gets the max time_point rather than one already in + // the past; this backstop only bounds the wait when the caller actually asked for a bound. + auto deadline = + options_.response_timeout_ > 0 + ? std::chrono::steady_clock::now() + std::chrono::seconds(options_.response_timeout_) + : std::chrono::steady_clock::time_point::max(); session->SendRequest(handler); // Wait for the response to be received @@ -478,10 +506,23 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] waiting for response from Elasticsearch (timeout = " << options_.response_timeout_ << " seconds)"); } - bool write_successful = handler->waitForResponse(); - - // End the session - session->FinishSession(); + bool timed_out = false; + bool write_successful = handler->waitForResponse(deadline, &timed_out); + + // Cancel only when the deadline genuinely expired with the transfer still outstanding: + // FinishSession() waits for an in-flight transfer to complete, which is exactly the hang + // this deadline exists to bound for HTTP clients (e.g. curl) whose worker thread blocks on + // the transfer itself. A terminal failure (ConnectFailed, SendFailed, CreateFailed, a + // response, ...) means the transfer is already over by the time waitForResponse returns, so + // FinishSession() is the correct call there, and for curl the two are not interchangeable. + if (timed_out) + { + session->CancelSession(); + } + else + { + session->FinishSession(); + } // If an error occurred with the HTTP request if (!write_successful) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 52a5d9b6ef..188a3af92d 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -22,6 +22,7 @@ #include #include #include +#include #include #include "nlohmann/json.hpp" @@ -120,6 +121,150 @@ class FakeHttpClient final : public http_client::HttpClient void SetMaxSessionsPerConnection(std::size_t) noexcept override {} }; +// A session that answers on a background thread after a short, fixed delay, so the exporter's +// wait genuinely blocks for a moment rather than finding the response already recorded: a +// response_timeout_ of 0 that mapped to a deadline already in the past would still pass a wait +// whose predicate is satisfied before the wait is even entered, so this is the only way to +// exercise a deadline comparison at all. +class DelayedSession final : public http_client::Session +{ +public: + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + + void SendRequest(std::shared_ptr handler) noexcept override + { + std::thread([handler] { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + FakeResponse response; + handler->OnResponse(response); + }).detach(); + } + + bool IsSessionActive() noexcept override { return true; } + bool CancelSession() noexcept override { return true; } + bool FinishSession() noexcept override { return true; } +}; + +class DelayedHttpClient final : public http_client::HttpClient +{ +public: + std::shared_ptr CreateSession( + opentelemetry::nostd::string_view) noexcept override + { + return std::make_shared(); + } + + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} +}; + +// A session that accepts a handler and never calls back into it, at all: no OnResponse, no +// OnEvent. Nothing in the HttpClient interface promises a terminal event, so a client built +// this way (a dead thread, a reused socket, a swallowed error) is a legal implementation, not +// a broken one. The exporter's own wait has to have a backstop independent of this. +// +// Only meaningful against the synchronous Export() path: under ENABLE_ASYNC_EXPORT, Export() +// hands the request to the client and returns success without waiting on anything, by design, +// so a silent client changes nothing observable there. +#ifndef ENABLE_ASYNC_EXPORT +class SilentSession final : public http_client::Session +{ +public: + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + + void SendRequest(std::shared_ptr) noexcept override {} + + bool IsSessionActive() noexcept override { return true; } + bool CancelSession() noexcept override + { + cancel_called_ = true; + return true; + } + bool FinishSession() noexcept override + { + finish_called_ = true; + return true; + } + + bool cancel_called_ = false; + bool finish_called_ = false; +}; + +class SilentHttpClient final : public http_client::HttpClient +{ +public: + std::shared_ptr CreateSession( + opentelemetry::nostd::string_view) noexcept override + { + session_ = std::make_shared(); + return session_; + } + + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + + std::shared_ptr session_; +}; + +// A session that delivers a prompt terminal failure event instead of never responding, so a +// case can distinguish "the transfer is already over" from "the deadline expired with nothing +// having reaped it yet". ConnectFailed stands in for any of the terminal failure events +// (SendFailed, CreateFailed, ...) that OnEvent() records as CompletionState::Failure. +class FailFastSession final : public http_client::Session +{ +public: + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + + void SendRequest(std::shared_ptr handler) noexcept override + { + handler->OnEvent(http_client::SessionState::ConnectFailed, ""); + } + + bool IsSessionActive() noexcept override { return true; } + bool CancelSession() noexcept override + { + cancel_called_ = true; + return true; + } + bool FinishSession() noexcept override + { + finish_called_ = true; + return true; + } + + bool cancel_called_ = false; + bool finish_called_ = false; +}; + +class FailFastHttpClient final : public http_client::HttpClient +{ +public: + std::shared_ptr CreateSession( + opentelemetry::nostd::string_view) noexcept override + { + session_ = std::make_shared(); + return session_; + } + + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + + std::shared_ptr session_; +}; +#endif // !ENABLE_ASYNC_EXPORT + } // namespace namespace sdklogs = opentelemetry::sdk::logs; @@ -157,6 +302,106 @@ TEST(ElasticsearchLogsExporterTests, CustomClientConstructionSucceeds) ASSERT_NE(exporter, nullptr); } +// Regression test: the synchronous export path used to wait on its response condition variable +// with no deadline of its own, trusting the injected HttpClient to eventually deliver a terminal +// event. SilentHttpClient never does, by design, so before the fix this test would hang forever. +// The 1-second response_timeout_ keeps the test itself fast while still exercising the real +// deadline path end to end, rather than a mocked-out clock. +// +// Synchronous-path-only: under ENABLE_ASYNC_EXPORT, Export() never waits at all (it hands the +// request off and returns success unconditionally), so there is nothing here to regress against. +#ifndef ENABLE_ASYNC_EXPORT +TEST(ElasticsearchLogsExporterTests, ExportReturnsOnTimeoutWhenClientNeverResponds) +{ + logs_exporter::ElasticsearchExporterOptions options("localhost", 9200, "logs", + /*response_timeout=*/1); + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + auto record = exporter->MakeRecordable(); + record->SetBody("this export should time out, not hang"); + + auto result = exporter->Export(nostd::span>(&record, 1)); + + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); +} + +// Regression test: on a timed-out export, Export() used to call session->FinishSession() +// regardless of the outcome. A real HTTP client (e.g. curl) blocks its FinishSession() on +// the in-flight transfer completing, which is exactly the hang the deadline exists to avoid, +// so the timeout path must cancel the session instead of finishing it. +TEST(ElasticsearchLogsExporterTests, ExportCancelsSessionOnTimeoutInsteadOfFinishing) +{ + logs_exporter::ElasticsearchExporterOptions options("localhost", 9200, "logs", + /*response_timeout=*/1); + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + auto record = exporter->MakeRecordable(); + record->SetBody("this export should cancel its session, not finish it"); + + auto result = exporter->Export(nostd::span>(&record, 1)); + + ASSERT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); + ASSERT_NE(http_client->session_, nullptr); + EXPECT_TRUE(http_client->session_->cancel_called_); + EXPECT_FALSE(http_client->session_->finish_called_); +} + +// Regression test: the timeout-vs-finish branch used to key on whether the export succeeded, +// not on whether the deadline actually expired. A terminal failure event (ConnectFailed, +// SendFailed, CreateFailed, ...) also fails the export, but the transfer is already over by +// then, so it must still be handed back with FinishSession(), the same as a successful export; +// only a deadline that expires with nothing having reaped the transfer yet should cancel. +TEST(ElasticsearchLogsExporterTests, ExportFinishesSessionOnTerminalFailureInsteadOfCancelling) +{ + logs_exporter::ElasticsearchExporterOptions options("localhost", 9200, "logs", + /*response_timeout=*/30); + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + auto record = exporter->MakeRecordable(); + record->SetBody("this export fails fast and should finish its session, not cancel it"); + + auto start = std::chrono::steady_clock::now(); + auto result = exporter->Export(nostd::span>(&record, 1)); + auto elapsed = std::chrono::steady_clock::now() - start; + + ASSERT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); + ASSERT_NE(http_client->session_, nullptr); + EXPECT_TRUE(http_client->session_->finish_called_); + EXPECT_FALSE(http_client->session_->cancel_called_); + // Confirms the failure was reported promptly rather than by the 30s response_timeout_ + // expiring, which would also leave cancel_called_ true. + EXPECT_LT(elapsed, std::chrono::seconds(1)); +} + +// Regression test: response_timeout = 0 is documented to mean no timeout, matching curl's own +// CURLOPT_TIMEOUT_MS 0 semantics. Deriving the wait's deadline as now() + seconds(0) instead +// mapped that to a deadline already in the past, so every sync export failed immediately even +// when the client was about to answer successfully. DelayedSession answers on a separate +// thread after a short, genuine delay, so a deadline already elapsed and a deadline far in the +// future are actually distinguishable here. +TEST(ElasticsearchLogsExporterTests, ZeroResponseTimeoutMeansNoTimeout) +{ + logs_exporter::ElasticsearchExporterOptions options("localhost", 9200, "logs", + /*response_timeout=*/0); + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + auto record = exporter->MakeRecordable(); + record->SetBody("this export should wait for the delayed response, not time out instantly"); + + auto result = exporter->Export(nostd::span>(&record, 1)); + + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); +} +#endif // !ENABLE_ASYNC_EXPORT + // Attempt to write a log to an invalid host/port, test that the Export() returns failure TEST(DISABLED_ElasticsearchLogsExporterTests, InvalidEndpoint) {