From 86ff1f6cd8939b1feccf012557efde50bbdcca29 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 01:23:30 +0800 Subject: [PATCH 1/4] [BUG] Wait on a completion state in the Elasticsearch exporter waitForResponse() waited on a condition variable with no predicate and then returned a bool that nothing on the error paths ever set. OnResponse() set the flag under the mutex and notified after releasing it, so a notification delivered before the exporting thread reached cv_.wait() was lost and that thread blocked with nothing left to wake it. The OnEvent failure cases were worse: they notified without taking the mutex and without recording anything at all. A spurious wakeup also returned false, reporting a request that was still in flight as failed. The handler now records an explicit completion state under the mutex on both the success and the failure paths, first writer wins so a session destroyed after a good response does not overwrite it, and the wait is on a predicate over that state. SessionState::Destroyed also records a failure if nothing was recorded yet, so a session that ends without a response no longer leaves the waiter blocked forever. Checked with a two-thread harness: with the notification delivered before the waiter parks, the old shape blocks until the test is killed while the new one returns immediately. Fixes #4296 Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- .../src/es_log_record_exporter.cc | 66 ++++++++++++++----- 1 file changed, 51 insertions(+), 15 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 8cb7d8140c..f08c39c146 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -5,6 +5,7 @@ #include #include #include +#include #include #include // IWYU pragma: keep #include @@ -100,21 +101,24 @@ class ResponseHandler : public http_client::EventHandler << log_message); } - // Set the response_received_ flag to true and notify any threads waiting on this result - response_received_ = true; + // Record the outcome and notify any threads waiting on this result + recordCompletionLocked(CompletionState::Success); } cv_.notify_all(); } /** - * A method the user calls to block their thread until the response is received. The longest - * duration is the timeout of the request, set by SetTimeoutMs() + * 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. */ bool waitForResponse() { std::unique_lock lk(mutex_); - cv_.wait(lk); - return response_received_; + // 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; }); + return completion_ == CompletionState::Success; } /** @@ -135,20 +139,23 @@ class ResponseHandler : public http_client::EventHandler { case http_client::SessionState::CreateFailed: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to create session"); - cv_.notify_all(); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::Created: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session created"); break; case http_client::SessionState::Destroyed: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session destroyed"); + // Nothing else will arrive after this. If no outcome was recorded, the session ended + // without a response, so release the waiter rather than leaving it blocked forever. + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::Connecting: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connecting to peer"); break; case http_client::SessionState::ConnectFailed: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to connect to peer"); - cv_.notify_all(); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::Connected: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connected to peer"); @@ -158,22 +165,22 @@ class ResponseHandler : public http_client::EventHandler break; case http_client::SessionState::SendFailed: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to send request"); - cv_.notify_all(); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::Response: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Received response"); break; case http_client::SessionState::SSLHandshakeFailed: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed SSL Handshake"); - cv_.notify_all(); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::TimedOut: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request timed out"); - cv_.notify_all(); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::NetworkError: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Network error"); - cv_.notify_all(); + recordCompletion(CompletionState::Failure); break; case http_client::SessionState::ReadError: OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Read error"); @@ -183,18 +190,47 @@ class ResponseHandler : public http_client::EventHandler break; case http_client::SessionState::Cancelled: OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] (manually) cancelled"); - cv_.notify_all(); + recordCompletion(CompletionState::Failure); break; } } private: + enum class CompletionState : std::uint8_t + { + Pending, + Success, + Failure + }; + + /** + * 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) + { + { + std::unique_lock lk(mutex_); + recordCompletionLocked(state); + } + cv_.notify_all(); + } + + /// As recordCompletion(), for callers that already hold mutex_ and notify themselves. + void recordCompletionLocked(CompletionState state) + { + if (completion_ == CompletionState::Pending) + { + completion_ = state; + } + } + // Define a condition variable and mutex std::condition_variable cv_; std::mutex mutex_; - // Whether the response from Elasticsearch has been received - bool response_received_ = false; + // Whether the request has completed, and how + CompletionState completion_ = CompletionState::Pending; // A string to store the response body std::string body_ = ""; From 2bb5768c2f505de676217379406f4e980df19039 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 01:25:12 +0800 Subject: [PATCH 2/4] Add CHANGELOG entry for #4298 Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8baba4503c..969b64bf6c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,9 @@ Increment the: namespaces [#4303](https://github.com/open-telemetry/opentelemetry-cpp/pull/4303) +* [BUG] Wait on a completion state in the Elasticsearch exporter + [#4298](https://github.com/open-telemetry/opentelemetry-cpp/pull/4298) + * [CODE HEALTH] Move metrics storage test fixtures into anonymous namespace [#4286](https://github.com/open-telemetry/opentelemetry-cpp/pull/4286) From c4f47d18c3c964afcbc2efa16473fff8fe1f3e3b Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 02:11:13 +0800 Subject: [PATCH 3/4] Drop stdint.h now that cstdint is included IWYU reports stdint.h as unused once cstdint is there for the enum base type. The only fixed-width type in the file is uint8_t, which cstdint provides. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- exporters/elasticsearch/src/es_log_record_exporter.cc | 1 - 1 file changed, 1 deletion(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index f08c39c146..a87a514f01 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -1,7 +1,6 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -#include #include #include #include From fc4ff1350b839f9567f587209333dac7f790ba7d Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Sat, 25 Jul 2026 04:11:18 +0800 Subject: [PATCH 4/4] Re-run CI The gcc 14 abiv2 run failed only on ext.http.curl.BasicCurlHttpTests.ElegantQuitQuick, which this change does not touch, so re-running to confirm it is a flake. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>