diff --git a/CHANGELOG.md b/CHANGELOG.md index 819a741b73..8f65090bad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,9 @@ Increment the: * [CODE HEALTH] Move remaining API test helpers into anonymous namespaces [#4301](https://github.com/open-telemetry/opentelemetry-cpp/pull/4301) +* [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) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 8cb7d8140c..a87a514f01 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -1,10 +1,10 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -#include #include #include #include +#include #include #include // IWYU pragma: keep #include @@ -100,21 +100,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 +138,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 +164,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 +189,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_ = "";