diff --git a/CHANGELOG.md b/CHANGELOG.md index d1766079f9..709c950b43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,18 @@ Increment the: to compile standalone on newer standard library implementations. [#4574](https://github.com/open-telemetry/opentelemetry-cpp/pull/4574) +* [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. + [#4362](https://github.com/open-telemetry/opentelemetry-cpp/issues/4362) + +* [DOC] Fix and clarify the `StartSpanOptions` documentation + [#4526](https://github.com/open-telemetry/opentelemetry-cpp/pull/4526) + ## [1.29.0] 2026-09-13 * [RELEASE] Bump main branch to 1.29.0-dev (#4259) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index cd3b1bdbfd..b9a064c45c 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -110,15 +110,32 @@ 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, as a TimedOut session event. But that + * guarantee belongs to the injected HttpClient, not to this exporter: a client that accepts + * a handler and never delivers a terminal event (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 +487,10 @@ 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()). + auto deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(options_.response_timeout_); session->SendRequest(handler); // Wait for the response to be received @@ -478,10 +499,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..4fd47431eb 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -120,6 +120,109 @@ class FakeHttpClient final : public http_client::HttpClient 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 +260,84 @@ 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)); +} +#endif // !ENABLE_ASYNC_EXPORT + // Attempt to write a log to an invalid host/port, test that the Export() returns failure TEST(DISABLED_ElasticsearchLogsExporterTests, InvalidEndpoint) {